TidesDB.cs

// Copyright (C) TidesDB
//
// Original Author: Alex Gaetano Padula
//
// Licensed under the Mozilla Public License, v. 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.mozilla.org/en-US/MPL/2.0/
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

using System.Runtime.InteropServices;
using System.Text;
using TidesDB.Native;

namespace TidesDB;

/// <summary>
/// TidesDB database instance.
/// </summary>
public sealed class TidesDb : IDisposable
{
    private nint _handle;
    private bool _disposed;
    private nint _dbPathPtr;
    private nint _objStorePtr;
    private nint _objStoreConfigPtr;
    private nint _localCachePathPtr;

    private TidesDb(nint handle, nint dbPathPtr, nint objStorePtr, nint objStoreConfigPtr, nint localCachePathPtr)
    {
        _handle = handle;
        _dbPathPtr = dbPathPtr;
        _objStorePtr = objStorePtr;
        _objStoreConfigPtr = objStoreConfigPtr;
        _localCachePathPtr = localCachePathPtr;
    }

    /// <summary>
    /// Opens a TidesDB instance with the given configuration.
    /// </summary>
    /// <param name="config">The database configuration.</param>
    /// <returns>A new TidesDB instance.</returns>
    public static TidesDb Open(Config config)
    {
        var dbPathPtr = Marshal.StringToHGlobalAnsi(config.DbPath);
        var objStorePtr = nint.Zero;
        var objStoreConfigPtr = nint.Zero;
        var localCachePathPtr = nint.Zero;

        try
        {
            if (config.ObjectStoreConfig is { } osCfg)
            {
                objStorePtr = osCfg.ConnectorType switch
                {
                    ObjectStoreConnectorType.Filesystem =>
                        NativeMethods.tidesdb_objstore_fs_create(
                            osCfg.FsRootDir ?? throw new ArgumentException("FsRootDir is required for filesystem connector")),
                    ObjectStoreConnectorType.S3 =>
                        throw new NotSupportedException("S3 connector requires native tidesdb_objstore_s3_create which is not yet exposed in the C# binding. Use the C API directly or contribute S3 support."),
                    _ => throw new ArgumentException($"Unknown connector type: {osCfg.ConnectorType}")
                };

                if (objStorePtr == nint.Zero)
                    throw new TidesDBException(-1, "failed to create object store connector");

                var nativeOsCfg = NativeMethods.tidesdb_objstore_default_config();

                if (osCfg.LocalCachePath != null)
                {
                    localCachePathPtr = Marshal.StringToHGlobalAnsi(osCfg.LocalCachePath);
                    nativeOsCfg.LocalCachePath = localCachePathPtr;
                }

                nativeOsCfg.LocalCacheMaxBytes = (nuint)osCfg.LocalCacheMaxBytes;
                nativeOsCfg.CacheOnRead = osCfg.CacheOnRead ? 1 : 0;
                nativeOsCfg.CacheOnWrite = osCfg.CacheOnWrite ? 1 : 0;
                nativeOsCfg.MaxConcurrentUploads = osCfg.MaxConcurrentUploads;
                nativeOsCfg.MaxConcurrentDownloads = osCfg.MaxConcurrentDownloads;
                nativeOsCfg.MultipartThreshold = (nuint)osCfg.MultipartThreshold;
                nativeOsCfg.MultipartPartSize = (nuint)osCfg.MultipartPartSize;
                nativeOsCfg.SyncManifestToObject = osCfg.SyncManifestToObject ? 1 : 0;
                nativeOsCfg.ReplicateWal = osCfg.ReplicateWal ? 1 : 0;
                nativeOsCfg.WalUploadSync = osCfg.WalUploadSync ? 1 : 0;
                nativeOsCfg.WalSyncThresholdBytes = (nuint)osCfg.WalSyncThresholdBytes;
                nativeOsCfg.WalSyncOnCommit = osCfg.WalSyncOnCommit ? 1 : 0;
                nativeOsCfg.ReplicaMode = osCfg.ReplicaMode ? 1 : 0;
                nativeOsCfg.ReplicaSyncIntervalUs = osCfg.ReplicaSyncIntervalUs;
                nativeOsCfg.ReplicaReplayWal = osCfg.ReplicaReplayWal ? 1 : 0;

                objStoreConfigPtr = Marshal.AllocHGlobal(Marshal.SizeOf<NativeObjStoreConfig>());
                Marshal.StructureToPtr(nativeOsCfg, objStoreConfigPtr, false);
            }

            var nativeConfig = new NativeConfig
            {
                DbPath = dbPathPtr,
                NumFlushThreads = config.NumFlushThreads,
                NumCompactionThreads = config.NumCompactionThreads,
                LogLevel = (int)config.LogLevel,
                BlockCacheSize = (nuint)config.BlockCacheSize,
                MaxOpenSstables = (nuint)config.MaxOpenSstables,
                LogToFile = config.LogToFile ? 1 : 0,
                LogTruncationAt = (nuint)config.LogTruncationAt,
                MaxMemoryUsage = (nuint)config.MaxMemoryUsage,
                UnifiedMemtable = config.UnifiedMemtable ? 1 : 0,
                UnifiedMemtableWriteBufferSize = (nuint)config.UnifiedMemtableWriteBufferSize,
                UnifiedMemtableSkipListMaxLevel = config.UnifiedMemtableSkipListMaxLevel,
                UnifiedMemtableSkipListProbability = config.UnifiedMemtableSkipListProbability,
                UnifiedMemtableSyncMode = (int)config.UnifiedMemtableSyncMode,
                UnifiedMemtableSyncIntervalUs = config.UnifiedMemtableSyncIntervalUs,
                ObjectStore = objStorePtr,
                ObjectStoreConfig = objStoreConfigPtr,
                MaxConcurrentFlushes = config.MaxConcurrentFlushes
            };

            var result = NativeMethods.tidesdb_open(ref nativeConfig, out var dbHandle);
            if (result != 0)
                throw new TidesDBException(result, "failed to open database");

            return new TidesDb(dbHandle, dbPathPtr, objStorePtr, objStoreConfigPtr, localCachePathPtr);
        }
        catch
        {
            Marshal.FreeHGlobal(dbPathPtr);
            if (localCachePathPtr != nint.Zero) Marshal.FreeHGlobal(localCachePathPtr);
            if (objStoreConfigPtr != nint.Zero) Marshal.FreeHGlobal(objStoreConfigPtr);
            throw;
        }
    }

    /// <summary>
    /// Creates a new column family with the given configuration.
    /// </summary>
    /// <param name="name">The column family name.</param>
    /// <param name="config">The column family configuration.</param>
    public void CreateColumnFamily(string name, ColumnFamilyConfig? config = null)
    {
        ThrowIfDisposed();
        config ??= ColumnFamilyConfig.Default;

        var nativeConfig = CreateNativeColumnFamilyConfig(config);
        var result = NativeMethods.tidesdb_create_column_family(_handle, name, ref nativeConfig);
        TidesDBException.ThrowIfError(result, "failed to create column family");
    }

    /// <summary>
    /// Drops a column family and all associated data.
    /// </summary>
    /// <param name="name">The column family name.</param>
    public void DropColumnFamily(string name)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_drop_column_family(_handle, name);
        TidesDBException.ThrowIfError(result, "failed to drop column family");
    }

    /// <summary>
    /// Deletes a column family by pointer. Faster than DropColumnFamily when you already
    /// hold a ColumnFamily handle, as it skips the name lookup.
    /// </summary>
    /// <param name="cf">The column family to delete.</param>
    public void DeleteColumnFamily(ColumnFamily cf)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_delete_column_family(_handle, cf.Handle);
        TidesDBException.ThrowIfError(result, "failed to delete column family");
    }

    /// <summary>
    /// Clones a column family, creating a complete copy with a new name.
    /// </summary>
    /// <param name="sourceName">The source column family name.</param>
    /// <param name="destName">The destination column family name.</param>
    public void CloneColumnFamily(string sourceName, string destName)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_clone_column_family(_handle, sourceName, destName);
        TidesDBException.ThrowIfError(result, "failed to clone column family");
    }

    /// <summary>
    /// Renames a column family atomically.
    /// </summary>
    /// <param name="oldName">The current column family name.</param>
    /// <param name="newName">The new column family name.</param>
    public void RenameColumnFamily(string oldName, string newName)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_rename_column_family(_handle, oldName, newName);
        TidesDBException.ThrowIfError(result, "failed to rename column family");
    }

    /// <summary>
    /// Creates an on-disk backup of the database without blocking reads/writes.
    /// </summary>
    /// <param name="dir">The backup directory path. Must be non-existent or empty.</param>
    public void Backup(string dir)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_backup(_handle, dir);
        TidesDBException.ThrowIfError(result, "failed to backup database");
    }

    /// <summary>
    /// Creates a lightweight, near-instant snapshot of the database using hard links.
    /// </summary>
    /// <param name="checkpointDir">The checkpoint directory path. Must be non-existent or empty.</param>
    public void Checkpoint(string checkpointDir)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_checkpoint(_handle, checkpointDir);
        TidesDBException.ThrowIfError(result, "failed to checkpoint database");
    }

    /// <summary>
    /// Gets a column family by name.
    /// </summary>
    /// <param name="name">The column family name.</param>
    /// <returns>The column family, or null if not found.</returns>
    public ColumnFamily? GetColumnFamily(string name)
    {
        ThrowIfDisposed();
        var cfHandle = NativeMethods.tidesdb_get_column_family(_handle, name);
        return cfHandle == nint.Zero ? null : new ColumnFamily(cfHandle);
    }

    /// <summary>
    /// Lists all column families in the database.
    /// </summary>
    /// <returns>An array of column family names.</returns>
    public string[] ListColumnFamilies()
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_list_column_families(_handle, out var namesPtr, out var count);
        TidesDBException.ThrowIfError(result, "failed to list column families");

        if (count == 0)
        {
            return [];
        }

        var names = new string[count];
        for (int i = 0; i < count; i++)
        {
            var namePtr = Marshal.ReadIntPtr(namesPtr, i * nint.Size);
            names[i] = Marshal.PtrToStringAnsi(namePtr) ?? "";
            NativeMethods.tidesdb_free(namePtr);
        }
        NativeMethods.tidesdb_free(namesPtr);

        return names;
    }

    /// <summary>
    /// Begins a new transaction with default isolation level.
    /// </summary>
    /// <returns>A new transaction.</returns>
    public Transaction BeginTransaction()
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_txn_begin(_handle, out var txnHandle);
        TidesDBException.ThrowIfError(result, "failed to begin transaction");
        return new Transaction(txnHandle);
    }

    /// <summary>
    /// Begins a new transaction with the specified isolation level.
    /// </summary>
    /// <param name="isolation">The isolation level.</param>
    /// <returns>A new transaction.</returns>
    public Transaction BeginTransaction(IsolationLevel isolation)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_txn_begin_with_isolation(_handle, (int)isolation, out var txnHandle);
        TidesDBException.ThrowIfError(result, "failed to begin transaction with isolation");
        return new Transaction(txnHandle);
    }

    /// <summary>
    /// Registers a custom comparator with the database.
    /// </summary>
    /// <param name="name">The comparator name.</param>
    /// <param name="ctxStr">Optional context string.</param>
    public void RegisterComparator(string name, string? ctxStr = null)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_register_comparator(_handle, name, nint.Zero, ctxStr, nint.Zero);
        TidesDBException.ThrowIfError(result, "failed to register comparator");
    }

    /// <summary>
    /// Retrieves a previously registered comparator by name.
    /// Returns true if the comparator is registered, false otherwise.
    /// </summary>
    /// <param name="name">The comparator name.</param>
    /// <returns>True if the comparator is registered.</returns>
    public bool GetComparator(string name)
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_get_comparator(_handle, name, out _, out _);
        return result == 0;
    }

    /// <summary>
    /// Promotes a read-only replica to primary mode, enabling writes.
    /// Performs a final MANIFEST sync and WAL replay before switching.
    /// </summary>
    public void PromoteToPrimary()
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_promote_to_primary(_handle);
        TidesDBException.ThrowIfError(result, "failed to promote to primary");
    }

    /// <summary>
    /// Forces a synchronous flush and aggressive compaction for all column families,
    /// then drains both the global flush and compaction queues. Blocks until complete.
    /// </summary>
    public void Purge()
    {
        ThrowIfDisposed();
        var result = NativeMethods.tidesdb_purge(_handle);
        TidesDBException.ThrowIfError(result, "failed to purge database");
    }

    /// <summary>
    /// Gets aggregate statistics across the entire database instance.
    /// </summary>
    /// <returns>Database-level statistics.</returns>
    public DbStats GetDbStats()
    {
        ThrowIfDisposed();
        var nativeStats = new NativeDbStats();
        var result = NativeMethods.tidesdb_get_db_stats(_handle, ref nativeStats);
        TidesDBException.ThrowIfError(result, "failed to get database stats");

        return new DbStats
        {
            NumColumnFamilies = nativeStats.NumColumnFamilies,
            TotalMemory = nativeStats.TotalMemory,
            AvailableMemory = nativeStats.AvailableMemory,
            ResolvedMemoryLimit = (ulong)nativeStats.ResolvedMemoryLimit,
            MemoryPressureLevel = nativeStats.MemoryPressureLevel,
            FlushPendingCount = nativeStats.FlushPendingCount,
            TotalMemtableBytes = nativeStats.TotalMemtableBytes,
            TotalImmutableCount = nativeStats.TotalImmutableCount,
            TotalSstableCount = nativeStats.TotalSstableCount,
            TotalDataSizeBytes = nativeStats.TotalDataSizeBytes,
            NumOpenSstables = nativeStats.NumOpenSstables,
            GlobalSeq = nativeStats.GlobalSeq,
            TxnMemoryBytes = nativeStats.TxnMemoryBytes,
            CompactionQueueSize = (ulong)nativeStats.CompactionQueueSize,
            FlushQueueSize = (ulong)nativeStats.FlushQueueSize,
            UnifiedMemtableEnabled = nativeStats.UnifiedMemtableEnabled != 0,
            UnifiedMemtableBytes = nativeStats.UnifiedMemtableBytes,
            UnifiedImmutableCount = nativeStats.UnifiedImmutableCount,
            UnifiedIsFlushing = nativeStats.UnifiedIsFlushing != 0,
            UnifiedNextCfIndex = nativeStats.UnifiedNextCfIndex,
            UnifiedWalGeneration = nativeStats.UnifiedWalGeneration,
            ObjectStoreEnabled = nativeStats.ObjectStoreEnabled != 0,
            ObjectStoreConnector = nativeStats.ObjectStoreConnector != nint.Zero
                ? Marshal.PtrToStringAnsi(nativeStats.ObjectStoreConnector) : null,
            LocalCacheBytesUsed = (ulong)nativeStats.LocalCacheBytesUsed,
            LocalCacheBytesMax = (ulong)nativeStats.LocalCacheBytesMax,
            LocalCacheNumFiles = nativeStats.LocalCacheNumFiles,
            LastUploadedGeneration = nativeStats.LastUploadedGeneration,
            UploadQueueDepth = (ulong)nativeStats.UploadQueueDepth,
            TotalUploads = nativeStats.TotalUploads,
            TotalUploadFailures = nativeStats.TotalUploadFailures,
            ReplicaMode = nativeStats.ReplicaMode != 0
        };
    }

    /// <summary>
    /// Gets statistics about the block cache.
    /// </summary>
    /// <returns>Cache statistics.</returns>
    public CacheStats GetCacheStats()
    {
        ThrowIfDisposed();
        var nativeStats = new NativeCacheStats();
        var result = NativeMethods.tidesdb_get_cache_stats(_handle, ref nativeStats);
        TidesDBException.ThrowIfError(result, "failed to get cache stats");

        return new CacheStats
        {
            Enabled = nativeStats.Enabled != 0,
            TotalEntries = (ulong)nativeStats.TotalEntries,
            TotalBytes = (ulong)nativeStats.TotalBytes,
            Hits = nativeStats.Hits,
            Misses = nativeStats.Misses,
            HitRate = nativeStats.HitRate,
            NumPartitions = (ulong)nativeStats.NumPartitions
        };
    }

    internal static unsafe NativeColumnFamilyConfig CreateNativeColumnFamilyConfigPublic(ColumnFamilyConfig config)
        => CreateNativeColumnFamilyConfig(config);

    private static unsafe NativeColumnFamilyConfig CreateNativeColumnFamilyConfig(ColumnFamilyConfig config)
    {
        var nativeConfig = new NativeColumnFamilyConfig
        {
            WriteBufferSize = (nuint)config.WriteBufferSize,
            LevelSizeRatio = (nuint)config.LevelSizeRatio,
            MinLevels = config.MinLevels,
            DividingLevelOffset = config.DividingLevelOffset,
            KlogValueThreshold = (nuint)config.KlogValueThreshold,
            CompressionAlgo = (int)config.CompressionAlgorithm,
            EnableBloomFilter = config.EnableBloomFilter ? 1 : 0,
            BloomFpr = config.BloomFpr,
            EnableBlockIndexes = config.EnableBlockIndexes ? 1 : 0,
            IndexSampleRatio = config.IndexSampleRatio,
            BlockIndexPrefixLen = config.BlockIndexPrefixLen,
            SyncMode = (int)config.SyncMode,
            SyncIntervalUs = config.SyncIntervalUs,
            SkipListMaxLevel = config.SkipListMaxLevel,
            SkipListProbability = config.SkipListProbability,
            DefaultIsolationLevel = (int)config.DefaultIsolationLevel,
            MinDiskSpace = config.MinDiskSpace,
            L1FileCountTrigger = config.L1FileCountTrigger,
            L0QueueStallThreshold = config.L0QueueStallThreshold,
            TombstoneDensityTrigger = config.TombstoneDensityTrigger,
            TombstoneDensityMinEntries = config.TombstoneDensityMinEntries,
            UseBtree = config.UseBtree ? 1 : 0,
            CommitHookFn = nint.Zero,
            CommitHookCtx = nint.Zero,
            ObjectTargetFileSize = 0,
            ObjectLazyCompaction = config.ObjectLazyCompaction ? 1 : 0,
            ObjectPrefetchCompaction = config.ObjectPrefetchCompaction ? 1 : 0,
            ComparatorFnCached = nint.Zero,
            ComparatorCtxCached = nint.Zero
        };

        if (!string.IsNullOrEmpty(config.ComparatorName))
        {
            var nameBytes = Encoding.UTF8.GetBytes(config.ComparatorName);
            var copyLen = Math.Min(nameBytes.Length, 63);
            for (int i = 0; i < copyLen; i++)
            {
                nativeConfig.ComparatorName[i] = nameBytes[i];
            }
            nativeConfig.ComparatorName[copyLen] = 0;
        }

        return nativeConfig;
    }

    private void ThrowIfDisposed()
    {
        ObjectDisposedException.ThrowIf(_disposed, this);
    }

    public void Dispose()
    {
        if (_disposed) return;
        _disposed = true;

        if (_handle != nint.Zero)
        {
            NativeMethods.tidesdb_close(_handle);
            _handle = nint.Zero;
        }

        if (_dbPathPtr != nint.Zero)
        {
            Marshal.FreeHGlobal(_dbPathPtr);
            _dbPathPtr = nint.Zero;
        }

        if (_localCachePathPtr != nint.Zero)
        {
            Marshal.FreeHGlobal(_localCachePathPtr);
            _localCachePathPtr = nint.Zero;
        }

        if (_objStoreConfigPtr != nint.Zero)
        {
            Marshal.FreeHGlobal(_objStoreConfigPtr);
            _objStoreConfigPtr = nint.Zero;
        }
    }
}