record method

Future<String> record({ required DVRecordTable table, required DVCaptureOp operation, required Object key, required int version, required Map<String, Object?> values, String? tenant, bool erased = false, })

Captures one write. Called by DVRecordTable after the row is written.

Inside DV.transaction the change is staged and published after commit; a rollback removes it unpublished, so no consumer can read it while the transaction is open nor after it has been undone. Outside a transaction it is published at once. Throws only when the change could not be written at all, so the caller can undo the write.

Implementation

Future<String> record({
  required DVRecordTable table,
  required DVCaptureOp operation,
  required Object key,
  required int version,
  required Map<String, Object?> values,
  String? tenant,
  bool erased = false,
}) async {
  await _ensureReady();
  track(table);
  final DVContext? transaction = DVTransactionRunner.activeContext;
  final String id = _newId('chg');
  final bool carriesRow =
      operation != DVCaptureOp.delete && operation != DVCaptureOp.snapshot;
  // Sensitive fields are left out here, before the log is written: a log
  // that holds a value and trusts every consumer to drop it has already
  // leaked it to its own storage and its backups.
  final Map<String, Object?>? row = carriesRow
      ? <String, Object?>{
          for (final String column in table.columns)
            if (!table.sensitive.contains(column))
              column: _jsonSafe(values[column]),
        }
      : null;
  final String? resolvedTenant = tenant ??
      (DVTenants.hasScope ? const DVTenants().currentTenant : null);
  final DateTime now = _clock();
  await _records.insert(logTable, <String, Object?>{
    'change_id': id,
    'change_seq': null,
    'model': table.table,
    'record_key': jsonEncode(_jsonSafe(key)),
    'operation': operation.name,
    'record_version': version,
    'tenant': resolvedTenant,
    'transaction_id': transaction?.transactionId,
    'occurred_at': _stamp(now),
    'write_order': _writeOrder(now),
    'published_at': null,
    'row_values': row == null ? null : jsonEncode(row),
    'redacted': jsonEncode(table.sensitive.toList()..sort()),
    'erased': erased ? 1 : 0,
    'purged': 0,
  });

  if (transaction == null) {
    await _publishQuietly(<String>[id]);
    return id;
  }

  final String transactionId = transaction.transactionId;
  final List<String>? staged = _staged[transactionId];
  if (staged == null) {
    _staged[transactionId] = <String>[id];
    transaction.afterCommit(() async {
      final List<String> ids = _staged.remove(transactionId) ?? <String>[];
      await _publishQuietly(ids);
    });
  } else {
    staged.add(id);
  }
  transaction.compensate(() async {
    final List<String>? open = _staged[transactionId];
    open?.remove(id);
    if (open != null && open.isEmpty) _staged.remove(transactionId);
    await _records.delete(
      logTable,
      where: DVFilter.all(<DVFilter>[
        DVFilter.equals('change_id', id),
        const DVFilter.isNull('change_seq'),
      ]),
    );
  });
  return id;
}