database.dart

import 'dart:async';
import 'dart:ffi' as ffi;
import 'dart:isolate';
import 'dart:io' show File, Platform;
import 'dart:typed_data';

import 'package:ffi/ffi.dart';
import 'package:resqlite/src/transaction.dart';
import 'package:resqlite/src/writer/write_worker.dart';
import 'package:resqlite/src/writer/writer.dart';

import 'diagnostics.dart';
import 'exceptions.dart';
import 'extensions/extension.dart';
import 'native/resqlite_bindings.dart';
import 'profile_counters.dart';
import 'profile_mode.dart';
import 'reader/reader_pool.dart';
import 'stream_engine.dart';
import 'tracelite_profile.dart';

part 'extensions/registration.dart';
part 'native/open_database.dart';

/// The result of [Database.selectBytes]: the JSON-encoded result [bytes] and
/// the [rowCount] of rows serialized into them.
///
/// [rowCount] is counted in C during the same serialization pass that produces
/// [bytes], so reading it is free — there is no second `COUNT(*)` query and no
/// need to parse the bytes to learn how many rows came back.
final class BytesResult {
  const BytesResult(this.bytes, this.rowCount);

  /// The query result as a single UTF-8 JSON array, serialized natively in C.
  final Uint8List bytes;

  /// The number of rows serialized into [bytes].
  final int rowCount;
}

/// A high-performance SQLite database with reactive queries.
///
/// All reads, writes, and reactive re-queries run off the main isolate
/// on persistent worker isolates, keeping your UI thread free.
///
/// ```dart
/// final db = await Database.open('app.db');
///
/// final rows = await db.select('SELECT * FROM users WHERE active = ?', [1]);
/// await db.execute('INSERT INTO users(name) VALUES (?)', ['Ada']);
///
/// db.stream('SELECT * FROM users').listen((users) {
///   print('${users.length} users');
/// });
///
/// await db.close();
/// ```
///
/// See also:
///
/// - [Transaction], for multi-statement atomic writes with read visibility
/// - [StreamEngine], for the reactive query lifecycle internals
final class Database {
  Database._(this._handle, this._path, int readerCount) {
    _runtime = Future.sync(() async {
      // Spawn the reader pool.
      final readerPool = await ReaderPool.spawn(_handle.address, readerCount);

      // Start the reactive query stream engine.
      final streamEngine = StreamEngine(readerPool);

      // Spawn the single writer isolate.
      final writer = await Writer.spawn(streamEngine, _handle);

      return (
        readerPool: readerPool,
        streamEngine: streamEngine,
        writer: writer,
      );
    });
  }

  final ffi.Pointer<ffi.Void> _handle;

  late final Future<_DatabaseRuntime> _runtime;

  /// The filesystem path the database was opened with. Retained so
  /// [diagnostics] can read the `-wal` sidecar size. `:memory:` or
  /// other non-file paths are stored verbatim; [diagnostics] detects
  /// and handles them.
  final String _path;

  Completer<void>? _closedCompleter = null;

  /// The raw native database handle.
  ///
  /// Exposed for advanced FFI interop only. Most applications should not
  /// need this.
  ffi.Pointer<ffi.Void> get handle => _handle;

  // -------------------------------------------------------------------------
  // Lifecycle
  // -------------------------------------------------------------------------

  void _ensureOpen() {
    if (_closedCompleter != null) {
      throw ResqliteConnectionException.databaseClosed();
    }
  }

  /// Opens or creates a SQLite database at [path].
  ///
  /// ```dart
  /// final db = await Database.open('app.db');
  /// ```
  ///
  /// If the file at [path] does not exist, a new database is created.
  /// Native open work runs on a temporary isolate. Reader and writer isolates
  /// are spawned non-blocking after that — the first query awaits their
  /// readiness automatically.
  ///
  /// If [encryptionKey] is provided, the database is encrypted using
  /// SQLite3 Multiple Ciphers (AES-256). The key must be a hex-encoded
  /// string (64 hex chars for a 256-bit key). All connections (writer +
  /// reader pool) use the same key.
  ///
  /// ```dart
  /// final db = await Database.open(
  ///   'secure.db',
  ///   encryptionKey: '0123456789abcdef0123456789abcdef'
  ///       '0123456789abcdef0123456789abcdef',
  /// );
  /// ```
  ///
  /// Throws a [ResqliteConnectionException] if the file cannot be opened
  /// or the encryption key is incorrect.
  ///
  /// The returned [Database] must be closed with [close] when no longer
  /// needed to release native resources.
  static Future<Database> open(
    String path, {
    String? encryptionKey,
    Iterable<ResqliteExtension> extensions = const [],
  }) async {
    // Determine the number of reader isolates to spawn.
    // cores - 1: leave one core for the main isolate (UI thread in Flutter).
    // min 2: so one worker sacrifice doesn't leave zero capacity.
    // max 4: benchmarked 2/4/8/16 workers
    // ([EXP-105](../../experiments/105-reader-pool-sizing.md)).
    // Concurrent query
    //   throughput plateaus at 4; raising past that regresses A11c
    //   many-streams-writer-throughput by ~55% and high-cardinality
    //   stream fan-out (A11b) by ~88% because each completed
    //   selectIfChanged reply queues another microtask ahead of the
    //   next pending write. Each idle worker costs ~30KB + one C
    //   reader connection.
    final readerCount = (Platform.numberOfProcessors - 1).clamp(2, 4);

    final handle = await _openNativeDatabase(
      path: path,
      encryptionKey: encryptionKey,
      readerCount: readerCount,
      extensions: extensions,
    );

    return Database._(handle, path, readerCount);
  }

  /// Closes this database, shutting down all worker isolates and releasing
  /// native resources.
  ///
  /// After `close()` resolves, any further operations on this [Database]
  /// throw a [ResqliteConnectionException].
  Future<void> close() async {
    if (_closedCompleter case Completer<void> completer) {
      return completer.future;
    }

    final completer = _closedCompleter = Completer<void>();

    try {
      final _DatabaseRuntime(:readerPool, :streamEngine, :writer) =
          await _runtime;

      streamEngine.close();
      await readerPool.close();
      await writer.close();

      resqliteClose(_handle);

      completer.complete();
    } catch (e) {
      completer.completeError(e);
      rethrow;
    }
  }

  // -------------------------------------------------------------------------
  // Read operations
  // -------------------------------------------------------------------------

  /// Runs a query and returns all matching rows.
  ///
  /// ```dart
  /// final users = await db.select(
  ///   'SELECT id, name FROM users WHERE active = ?',
  ///   [1],
  /// );
  /// for (final user in users) {
  ///   print('${user['id']}: ${user['name']}');
  /// }
  /// ```
  ///
  /// The [parameters] list is bound positionally to `?` placeholders in
  /// [sql]. Returns an empty list if no rows match.
  ///
  /// The returned rows are lightweight [Row] views over a shared result
  /// buffer — accessing `row['column']` is a hash lookup, not a map copy.
  /// Use `Map<String, Object?>.from(row)` if you need a mutable copy.
  ///
  /// Runs on a background worker isolate. The main isolate only receives
  /// the finished result.
  ///
  /// Throws a [ResqliteQueryException] if the SQL is malformed.
  ///
  /// See also:
  ///
  /// - [selectBytes], for JSON-encoded results without Dart object allocation
  /// - [stream], for reactive queries that re-emit on writes
  Future<List<Map<String, Object?>>> select(
    String sql, [
    List<Object?> parameters = const [],
  ]) async {
    final transaction = Transaction.current;
    if (transaction != null) {
      return transaction.select(sql, parameters);
    }

    _ensureOpen();

    // No post-await _ensureOpen re-check: if close() has run while we
    // were parked, the pool itself now rejects dispatch with
    // ResqliteConnectionException (see ReaderPool._dispatch). That lets
    // *in-flight* reads that had already dispatched to a worker finish
    // via the pool's drain semantics, while reads still parked on the
    // pool future bail out cleanly.
    final _DatabaseRuntime(:readerPool) = await _runtime;
    final int? correlationId = kProfileMode && kTraceliteProfileMode
        ? TraceliteProfile.nextCorrelationId()
        : null;
    if (correlationId == null) {
      return readerPool.select(sql, parameters);
    }
    final sqlId = TraceliteProfile.internString(sql);
    return TraceliteProfile.traceAsync(
      TraceliteResqliteSpans.databaseSelect,
      () => readerPool.select(sql, parameters, correlationId),
      correlationId: correlationId,
      beginArgs: [sqlId, parameters.length],
      endArgs: (rows) => [rows.length],
    );
  }

  /// Executes a query and returns the result as JSON-encoded bytes plus the
  /// number of rows serialized into them (see [BytesResult]).
  ///
  /// ```dart
  /// final result = await db.selectBytes(
  ///   'SELECT id, name FROM users WHERE active = ?',
  ///   [1],
  /// );
  /// // result.bytes is a Uint8List containing a JSON array, e.g.:
  /// // [{"id":1,"name":"Ada"},{"id":2,"name":"Grace"}]
  /// // result.rowCount is 2.
  /// ```
  ///
  /// JSON serialization happens entirely in C — no Dart [Map] or [String]
  /// objects are created for the result data. The bytes cross to Dart as a
  /// single [Uint8List]. [BytesResult.rowCount] is counted during that same
  /// C pass, so it costs nothing to read and saves callers a `COUNT(*)` or a
  /// parse when they need the row count (e.g. to build a paging envelope).
  ///
  /// This is ideal for HTTP responses, file export, or any path where the
  /// end consumer wants JSON bytes rather than Dart objects.
  ///
  /// **Note:** This method always reads from the reader pool, even inside
  /// a [transaction]. Use [select] if you need to see uncommitted writes.
  ///
  /// Throws a [ResqliteQueryException] if the SQL is malformed.
  /// Throws [StateError] if called inside a [transaction] body.
  Future<BytesResult> selectBytes(
    String sql, [
    List<Object?> parameters = const [],
  ]) async {
    if (Transaction.current != null) {
      throw StateError(
        'selectBytes() cannot be used inside a transaction. '
        'Use select() instead, which sees uncommitted writes.',
      );
    }

    _ensureOpen();

    final _DatabaseRuntime(:readerPool) = await _runtime;
    final int? correlationId = kProfileMode && kTraceliteProfileMode
        ? TraceliteProfile.nextCorrelationId()
        : null;
    final ({Uint8List bytes, int rowCount}) result;
    if (correlationId == null) {
      result = await readerPool.selectBytes(sql, parameters);
    } else {
      final sqlId = TraceliteProfile.internString(sql);
      result = await TraceliteProfile.traceAsync(
        TraceliteResqliteSpans.databaseSelectBytes,
        () => readerPool.selectBytes(sql, parameters, correlationId),
        correlationId: correlationId,
        beginArgs: [sqlId, parameters.length],
        endArgs: (r) => [r.bytes.length, r.rowCount],
      );
    }
    return BytesResult(result.bytes, result.rowCount);
  }

  // -------------------------------------------------------------------------
  // Reactive queries
  // -------------------------------------------------------------------------

  /// Creates a reactive query that re-emits results when underlying tables
  /// change.
  ///
  /// ```dart
  /// db.stream('SELECT * FROM tasks WHERE done = ?', [0]).listen((tasks) {
  ///   print('${tasks.length} open tasks');
  /// });
  /// ```
  ///
  /// The first emission contains the current results. Subsequent emissions
  /// occur after any write that modifies tables this query depends on.
  ///
  /// Table dependencies are detected automatically via SQLite's authorizer
  /// hook — works with table-backed JOINs, subqueries, views, and CTEs without
  /// requiring a manual table list.
  ///
  /// Direct virtual-table / FTS streams are a known limitation. SQLite's
  /// preupdate hook does not report virtual-table writes, so queries that only
  /// depend on a virtual table may not re-emit automatically. For
  /// external-content FTS, join the real content table in the streamed query so
  /// normal table invalidation can apply.
  ///
  /// Streams are deduplicated: multiple calls with the same [sql] and
  /// [parameters] share a single underlying query. New listeners on an
  /// existing stream receive the cached result immediately.
  ///
  /// Unchanged results are suppressed — if a write touches a watched table
  /// but doesn't change this query's output, no emission occurs.
  ///
  /// Create streams once and reuse them (e.g., as `late final` fields in
  /// a `State` class), rather than creating new streams on every build.
  Stream<List<Map<String, Object?>>> stream(
    String sql, [
    List<Object?> parameters = const [],
  ]) {
    _ensureOpen();
    return Stream.fromFuture(
      _runtime,
    ).asyncExpand((runtime) => runtime.streamEngine.stream(sql, parameters));
  }

  // -------------------------------------------------------------------------
  // Write operations
  // -------------------------------------------------------------------------

  /// Executes a write statement and returns the result.
  ///
  /// ```dart
  /// final result = await db.execute(
  ///   'INSERT INTO users(name, email) VALUES (?, ?)',
  ///   ['Ada', 'ada@example.com'],
  /// );
  /// print('Inserted row ${result.lastInsertId}');
  /// print('${result.affectedRows} row(s) affected');
  /// ```
  ///
  /// The [parameters] list is bound positionally to `?` placeholders in
  /// [sql]. Each element must be a [String], [int], [double], [Uint8List]
  /// (for blobs), or `null`.
  ///
  /// Suitable for INSERT, UPDATE, DELETE, and DDL statements. For queries
  /// that return rows, use [select] instead.
  ///
  /// Any active [stream] queries watching the affected tables are
  /// automatically re-queried after this write commits.
  ///
  /// Throws a [ResqliteQueryException] if the SQL is malformed or
  /// violates a constraint.
  Future<WriteResult> execute(
    String sql, [
    List<Object?> parameters = const [],
  ]) async {
    final transaction = Transaction.current;
    if (transaction != null) {
      return transaction.execute(sql, parameters);
    }

    _ensureOpen();

    final _DatabaseRuntime(:streamEngine, :writer) = await _runtime;
    final int? correlationId = kProfileMode && kTraceliteProfileMode
        ? TraceliteProfile.nextCorrelationId()
        : null;

    Future<ExecuteResponse> write() =>
        writer.execute(sql, parameters, correlationId);

    final ExecuteResponse response;
    if (correlationId == null) {
      response = await write();
    } else {
      final sqlId = TraceliteProfile.internString(sql);
      response = await TraceliteProfile.traceAsync(
        TraceliteResqliteSpans.databaseExecute,
        write,
        correlationId: correlationId,
        beginArgs: [sqlId, parameters.length],
        endArgs: (response) => [response.result.affectedRows],
      );
    }

    ProfileCounters.recordWriterSqlite(response.writerSqliteUs);
    streamEngine.onDependencyChanges(
      response.modifications,
      traceCorrelationId: correlationId,
    );

    return response.result;
  }

  /// Executes one SQL statement across many parameter sets in a single
  /// transaction.
  ///
  /// ```dart
  /// await db.executeBatch(
  ///   'INSERT INTO users(name) VALUES (?)',
  ///   [['Ada'], ['Grace'], ['Sonja']],
  /// );
  /// ```
  ///
  /// The statement is prepared once and reused across all [paramSets],
  /// wrapped in a single BEGIN/COMMIT transaction. This is significantly
  /// faster than calling [execute] in a loop.
  ///
  /// All-or-nothing: if any row fails, the entire batch rolls back.
  ///
  /// Streams watching the affected table fire once on commit, not per row.
  ///
  /// Throws a [ResqliteQueryException] if any statement fails.
  Future<void> executeBatch(String sql, List<List<Object?>> paramSets) async {
    final transaction = Transaction.current;
    if (transaction != null) {
      return transaction.executeBatch(sql, paramSets);
    }

    _ensureOpen();

    final _DatabaseRuntime(:streamEngine, :writer) = await _runtime;

    final int? correlationId = kProfileMode && kTraceliteProfileMode
        ? TraceliteProfile.nextCorrelationId()
        : null;

    Future<BatchResponse?> write() =>
        writer.executeBatch(sql, paramSets, traceCorrelationId: correlationId);

    final BatchResponse? response;
    if (correlationId == null) {
      response = await write();
    } else {
      final sqlId = TraceliteProfile.internString(sql);
      final paramCount = paramSets.isEmpty ? 0 : paramSets.first.length;
      response = await TraceliteProfile.traceAsync(
        TraceliteResqliteSpans.databaseExecuteBatch,
        write,
        correlationId: correlationId,
        beginArgs: [sqlId, paramCount, paramSets.length],
      );
    }

    if (response != null) {
      ProfileCounters.recordWriterSqlite(response.writerSqliteUs);
      streamEngine.onDependencyChanges(
        response.modifications,
        traceCorrelationId: correlationId,
      );
    }
  }

  /// Runs [body] inside a database transaction.
  ///
  /// ```dart
  /// final count = await db.transaction((tx) async {
  ///   await tx.execute('INSERT INTO users(name) VALUES (?)', ['Ada']);
  ///   final rows = await tx.select('SELECT COUNT(*) as c FROM users');
  ///   return rows.first['c'] as int;
  /// });
  /// ```
  ///
  /// All operations within [body] are applied atomically. If [body]
  /// completes normally, the transaction commits. If [body] throws,
  /// the transaction rolls back and the exception is rethrown.
  ///
  /// The [Transaction] passed to [body] supports both [Transaction.execute]
  /// and [Transaction.select]. Reads inside the transaction see uncommitted
  /// writes from earlier statements in the same transaction.
  ///
  /// Stream invalidation happens once on commit, not per statement.
  /// Rolled-back transactions do not trigger stream re-queries.
  ///
  /// Returns the value returned by [body].
  Future<T> transaction<T>(Future<T> Function(Transaction tx) body) async {
    final transaction = Transaction.current;
    if (transaction != null) {
      return transaction.transaction(body);
    }

    _ensureOpen();

    final runtime = await _runtime;
    final writer = runtime.writer;
    final int? correlationId = kProfileMode && kTraceliteProfileMode
        ? TraceliteProfile.nextCorrelationId()
        : null;
    Future<T> run() => writer.locked(
      () => writer.transaction(body, traceCorrelationId: correlationId),
    );
    if (correlationId == null) {
      return run();
    }
    return TraceliteProfile.traceAsync(
      TraceliteResqliteSpans.databaseTransaction,
      run,
      correlationId: correlationId,
    );
  }

  // -------------------------------------------------------------------------
  // Diagnostics
  // -------------------------------------------------------------------------

  /// Captures a per-connection diagnostic snapshot.
  ///
  /// Aggregates SQLite's `sqlite3_db_status` counters across the writer
  /// and any idle reader connections, plus a filesystem-level read of
  /// the `-wal` sidecar. Intended primarily for benchmark instrumentation
  /// and mobile memory reporting.
  ///
  /// ```dart
  /// final d = await db.diagnostics();
  /// print('page cache: ${d.sqlitePageCacheBytes} bytes');
  /// print('WAL size:   ${d.walBytes} bytes');
  /// ```
  ///
  /// If one or more reader connections are busy at snapshot time, their
  /// byte counters are excluded from the totals and
  /// [Diagnostics.readersBusyAtSnapshot] is set to `true`. Taking
  /// snapshots between operations (when no concurrent work is in flight)
  /// produces clean totals.
  ///
  /// See [Diagnostics] for field-level semantics.
  Future<Diagnostics> diagnostics() async {
    _ensureOpen();

    // Read the three SQLite counters. Each call returns the aggregate
    // across writer + idle readers; if any reader was busy, the C
    // layer has still populated the idle-subset aggregate into the
    // out pointers. `getDbStatusTotalAllowBusy` surfaces those partial
    // numbers with a `partial: true` flag instead of discarding them.
    var readersBusy = false;

    int readCounter(int op) {
      final r = getDbStatusTotalAllowBusy(_handle, op);
      if (r.partial) readersBusy = true;
      return r.current;
    }

    final pageCache = readCounter(SqliteDbStatusOp.cacheUsed);
    final schema = readCounter(SqliteDbStatusOp.schemaUsed);
    final stmt = readCounter(SqliteDbStatusOp.stmtUsed);

    // WAL sidecar — only meaningful for on-disk databases.
    var walBytes = 0;
    if (_path != ':memory:' && !_path.startsWith('file::memory:')) {
      try {
        walBytes = await File('$_path-wal').length();
      } on Object {
        // File may not exist yet (no writes since open, or fresh DB) or
        // path may be unreadable for some reason. Treat as zero; the
        // only caller is diagnostics, not a correctness path.
        walBytes = 0;
      }
    }

    final _DatabaseRuntime(:streamEngine) = await _runtime;
    // [EXP-183] Per-reader json_buf high-water. Safe to call concurrently
    // with reader activity: each reader's cap is an int and tearing only
    // widens an already-bounded summary number, fine for a diagnostic.
    final jsonBufTotal = resqliteReaderJsonBufTotal(_handle);
    return Diagnostics(
      sqlitePageCacheBytes: pageCache,
      sqliteSchemaBytes: schema,
      sqliteStmtBytes: stmt,
      walBytes: walBytes,
      readersBusyAtSnapshot: readersBusy,
      streamLength: streamEngine.length,
      unknownDependencyFallbackCount:
          streamEngine.unknownDependencyFallbackCount,
      readerJsonBufHighWaterBytes: jsonBufTotal,
    );
  }
}

typedef _DatabaseRuntime = ({
  ReaderPool readerPool,
  StreamEngine streamEngine,
  Writer writer,
});