diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedEventStore.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedEventStore.java
new file mode 100644
index 00000000..fb86b2d0
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedEventStore.java
@@ -0,0 +1,292 @@
+package io.teaql.core.sqlite;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.LongFunction;
+
+/**
+ * SQLite durability boundary for already-committed logical mutation facts.
+ *
+ * The caller must install the schema before use. Append and apply require the
+ * same active SQL transaction as business writes. This class deliberately does
+ * not infer mutations from a ledger, replay business logic, or expose a remote
+ * replication endpoint. Projection writers are trusted infrastructure code.
+ */
+public final class SqliteCommittedEventStore {
+ private static final ObjectMapper JSON = new ObjectMapper();
+ private static final java.util.Set ENVELOPE_FIELDS = java.util.Set.of(
+ "protocol_version", "change_set_id", "source_runtime_id", "source_sequence",
+ "source_model_hash", "source_transaction_id", "committed_at", "audit", "mutations");
+ private final SqlExecutionAdapter sql;
+
+ public SqliteCommittedEventStore(SqlExecutionAdapter sql) {
+ this.sql = Objects.requireNonNull(sql, "sql");
+ }
+
+ public record OutboxRecord(String sourceRuntimeId, long sourceSequence,
+ String changeSetId, String modelHash, String payloadSha256, String payloadJson) {
+ @Override public String toString() {
+ return "OutboxRecord[sourceRuntimeId=" + sourceRuntimeId
+ + ", sourceSequence=" + sourceSequence + ", payload=]";
+ }
+ }
+
+ public record InboxOutcome(boolean duplicate, long checkpoint) {
+ public static InboxOutcome applied(long checkpoint) { return new InboxOutcome(false, checkpoint); }
+ public static InboxOutcome duplicate(long checkpoint) { return new InboxOutcome(true, checkpoint); }
+ }
+
+ @FunctionalInterface
+ interface TrustedProjectionWriter {
+ void apply(SqlExecutionAdapter transaction, JsonNode committedFacts);
+ }
+
+ public static final class CommittedEventException extends RuntimeException {
+ private final String code;
+
+ CommittedEventException(String code, String message) {
+ super(message);
+ this.code = code;
+ }
+
+ public String code() { return code; }
+ }
+
+ public void ensureSchema() {
+ SqliteCommittedObservation.ensureSchema(sql);
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_sequence ("
+ + "source_runtime_id TEXT PRIMARY KEY, "
+ + "current_sequence INTEGER NOT NULL CHECK(current_sequence >= 0))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_outbox ("
+ + "source_runtime_id TEXT NOT NULL, "
+ + "source_sequence INTEGER NOT NULL CHECK(source_sequence > 0), "
+ + "change_set_id TEXT NOT NULL, model_hash TEXT NOT NULL, "
+ + "payload_sha256 TEXT NOT NULL, payload_json TEXT NOT NULL, "
+ + "PRIMARY KEY(source_runtime_id, source_sequence), "
+ + "UNIQUE(source_runtime_id, change_set_id))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_inbox ("
+ + "source_runtime_id TEXT NOT NULL, change_set_id TEXT NOT NULL, "
+ + "source_sequence INTEGER NOT NULL CHECK(source_sequence > 0), "
+ + "payload_sha256 TEXT NOT NULL, PRIMARY KEY(source_runtime_id, change_set_id))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_applied_payload ("
+ + "source_runtime_id TEXT NOT NULL, source_sequence INTEGER NOT NULL CHECK(source_sequence > 0), "
+ + "change_set_id TEXT NOT NULL, payload_json TEXT NOT NULL, "
+ + "PRIMARY KEY(source_runtime_id, source_sequence), "
+ + "UNIQUE(source_runtime_id, change_set_id))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_checkpoint ("
+ + "source_runtime_id TEXT PRIMARY KEY, "
+ + "source_sequence INTEGER NOT NULL CHECK(source_sequence >= 0), "
+ + "model_hash TEXT NOT NULL)");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_delivery_cursor ("
+ + "source_runtime_id TEXT NOT NULL, target_id TEXT NOT NULL, "
+ + "acked_sequence INTEGER NOT NULL CHECK(acked_sequence >= 0), "
+ + "model_hash TEXT NOT NULL, PRIMARY KEY(source_runtime_id, target_id))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_snapshot ("
+ + "source_runtime_id TEXT PRIMARY KEY, snapshot_id TEXT NOT NULL, "
+ + "source_sequence INTEGER NOT NULL CHECK(source_sequence >= 0), model_hash TEXT NOT NULL, "
+ + "payload_sha256 TEXT NOT NULL, payload_json TEXT NOT NULL)");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_snapshot_publication ("
+ + "source_runtime_id TEXT NOT NULL, target_id TEXT NOT NULL, snapshot_id TEXT NOT NULL, "
+ + "source_sequence INTEGER NOT NULL CHECK(source_sequence >= 0), model_hash TEXT NOT NULL, "
+ + "payload_sha256 TEXT NOT NULL, payload_json TEXT NOT NULL, "
+ + "acknowledged INTEGER NOT NULL CHECK(acknowledged IN (0,1)), "
+ + "PRIMARY KEY(source_runtime_id,target_id))");
+ }
+
+ /** Bounded publisher read of one durable, ordered source stream. */
+ public List outboxAfter(String sourceRuntimeId, long after, int limit) {
+ if (blank(sourceRuntimeId) || after < 0 || limit < 1 || limit > 1000) {
+ throw invalid("invalid source, position, or read limit");
+ }
+ return sql.queryForList("SELECT source_runtime_id, source_sequence, change_set_id, "
+ + "model_hash, payload_sha256, payload_json FROM teaql_event_outbox "
+ + "WHERE source_runtime_id = ? AND source_sequence > ? "
+ + "ORDER BY source_sequence ASC LIMIT ?", new Object[] {sourceRuntimeId, after, limit})
+ .stream().map(row -> new OutboxRecord((String) row.get("source_runtime_id"),
+ number(row.get("source_sequence")), (String) row.get("change_set_id"),
+ (String) row.get("model_hash"), (String) row.get("payload_sha256"),
+ (String) row.get("payload_json"))).toList();
+ }
+
+ /** Allocate and append within the caller's already-open business transaction. */
+ public OutboxRecord append(String sourceRuntimeId, String changeSetId, String modelHash,
+ LongFunction payloadBuilder) {
+ requireTransaction();
+ if (blank(sourceRuntimeId) || blank(changeSetId) || blank(modelHash) || payloadBuilder == null) {
+ throw invalid("missing committed-event identity or payload builder");
+ }
+ return savepoint("teaql_event_append", () -> {
+ sql.update("INSERT INTO teaql_event_sequence(source_runtime_id, current_sequence) "
+ + "VALUES (?, 0) ON CONFLICT(source_runtime_id) DO NOTHING",
+ new Object[] {sourceRuntimeId});
+ int updated = sql.update("UPDATE teaql_event_sequence SET current_sequence = current_sequence + 1 "
+ + "WHERE source_runtime_id = ? AND current_sequence < 9223372036854775807",
+ new Object[] {sourceRuntimeId});
+ if (updated != 1) throw new CommittedEventException("SEQUENCE_EXHAUSTED", "source sequence exhausted");
+ long sequence = number(sql.queryForList("SELECT current_sequence FROM teaql_event_sequence "
+ + "WHERE source_runtime_id = ?", new Object[] {sourceRuntimeId}).get(0).get("current_sequence"));
+ String payloadJson = payloadBuilder.apply(sequence);
+ JsonNode payload = parse(payloadJson);
+ validateHeader(payload, sourceRuntimeId, sequence, changeSetId, modelHash);
+ String digest = digest(payloadJson);
+ sql.update("INSERT INTO teaql_event_outbox "
+ + "(source_runtime_id, source_sequence, change_set_id, model_hash, payload_sha256, payload_json) "
+ + "VALUES (?, ?, ?, ?, ?, ?)",
+ new Object[] {sourceRuntimeId, sequence, changeSetId, modelHash, digest, payloadJson});
+ return new OutboxRecord(sourceRuntimeId, sequence, changeSetId, modelHash, digest, payloadJson);
+ });
+ }
+
+ /**
+ * Apply one set through a restricted projection writer. The writer must not
+ * call normal entity save, Checker/Fix, ID generation or business hooks.
+ */
+ InboxOutcome apply(OutboxRecord record, String expectedModelHash,
+ TrustedProjectionWriter writer) {
+ requireTransaction();
+ if (record == null || blank(expectedModelHash) || writer == null) {
+ throw invalid("missing committed event, model hash, or projection writer");
+ }
+ if (!expectedModelHash.equals(record.modelHash())) {
+ throw new CommittedEventException("MODEL_MISMATCH", "committed-event model hash mismatch");
+ }
+ JsonNode payload = parse(record.payloadJson());
+ validateHeader(payload, record.sourceRuntimeId(), record.sourceSequence(),
+ record.changeSetId(), record.modelHash());
+ if (!digest(record.payloadJson()).equals(record.payloadSha256())) {
+ throw invalid("payload digest mismatch");
+ }
+ List> checkpoints = sql.queryForList(
+ "SELECT source_sequence, model_hash FROM teaql_event_checkpoint WHERE source_runtime_id = ?",
+ new Object[] {record.sourceRuntimeId()});
+ long last = checkpoints.isEmpty() ? 0 : number(checkpoints.get(0).get("source_sequence"));
+ if (!checkpoints.isEmpty() && !expectedModelHash.equals(checkpoints.get(0).get("model_hash"))) {
+ throw new CommittedEventException("MODEL_MISMATCH", "committed-event checkpoint model mismatch");
+ }
+ List> prior = sql.queryForList(
+ "SELECT source_sequence, payload_sha256 FROM teaql_event_inbox "
+ + "WHERE source_runtime_id = ? AND change_set_id = ?",
+ new Object[] {record.sourceRuntimeId(), record.changeSetId()});
+ if (!prior.isEmpty()) {
+ Map seen = prior.get(0);
+ if (number(seen.get("source_sequence")) == record.sourceSequence()
+ && record.payloadSha256().equals(seen.get("payload_sha256"))) {
+ List> retained = sql.queryForList(
+ "SELECT source_sequence, payload_json FROM teaql_event_applied_payload "
+ + "WHERE source_runtime_id = ? AND change_set_id = ?",
+ new Object[] {record.sourceRuntimeId(), record.changeSetId()});
+ if (retained.size() != 1
+ || number(retained.get(0).get("source_sequence")) != record.sourceSequence()
+ || !record.payloadSha256().equals(digest((String) retained.get(0).get("payload_json")))) {
+ throw invalid("prior Inbox identity lacks retained source evidence");
+ }
+ if (last < record.sourceSequence()) {
+ throw invalid("prior Inbox identity is ahead of target checkpoint");
+ }
+ return InboxOutcome.duplicate(record.sourceSequence());
+ }
+ throw new CommittedEventException("DUPLICATE_CONFLICT", "committed-event identity has conflicting content");
+ }
+ if (last == Long.MAX_VALUE) {
+ throw new CommittedEventException("SEQUENCE_EXHAUSTED", "source sequence exhausted");
+ }
+ if (record.sourceSequence() != last + 1) {
+ throw new CommittedEventException("SEQUENCE_GAP",
+ "committed-event sequence gap: expected " + (last + 1) + ", received " + record.sourceSequence());
+ }
+ return savepoint("teaql_event_apply", () -> {
+ writer.apply(sql, payload);
+ sql.update("INSERT INTO teaql_event_applied_payload "
+ + "(source_runtime_id, source_sequence, change_set_id, payload_json) VALUES (?, ?, ?, ?)",
+ new Object[] {record.sourceRuntimeId(), record.sourceSequence(),
+ record.changeSetId(), record.payloadJson()});
+ sql.update("INSERT INTO teaql_event_inbox "
+ + "(source_runtime_id, change_set_id, source_sequence, payload_sha256) VALUES (?, ?, ?, ?)",
+ new Object[] {record.sourceRuntimeId(), record.changeSetId(),
+ record.sourceSequence(), record.payloadSha256()});
+ sql.update("INSERT INTO teaql_event_checkpoint (source_runtime_id, source_sequence, model_hash) "
+ + "VALUES (?, ?, ?) ON CONFLICT(source_runtime_id) DO UPDATE SET "
+ + "source_sequence = excluded.source_sequence, model_hash = excluded.model_hash",
+ new Object[] {record.sourceRuntimeId(), record.sourceSequence(), record.modelHash()});
+ return InboxOutcome.applied(record.sourceSequence());
+ });
+ }
+
+ private static JsonNode parse(String payloadJson) {
+ if (payloadJson == null) throw invalid("missing payload");
+ try {
+ return JSON.readTree(payloadJson);
+ } catch (Exception ignored) {
+ throw invalid("invalid payload JSON");
+ }
+ }
+
+ private static void validateHeader(JsonNode payload, String source, long sequence,
+ String changeSetId, String modelHash) {
+ if (blank(source) || sequence < 1 || blank(changeSetId) || blank(modelHash)
+ || payload == null || payload.path("protocol_version").asInt(-1) != 1
+ || !source.equals(payload.path("source_runtime_id").asText())
+ || !Long.toString(sequence).equals(payload.path("source_sequence").asText())
+ || !changeSetId.equals(payload.path("change_set_id").asText())
+ || !modelHash.equals(payload.path("source_model_hash").asText())
+ || !payload.path("mutations").isArray() || payload.path("mutations").isEmpty()) {
+ throw invalid("required v1 header or mutations mismatch");
+ }
+ rejectUnknownControls(payload, ENVELOPE_FIELDS);
+ }
+
+ /** Check protocol control frames, never recursively scan modeled values or audit text. */
+ static void rejectUnknownControls(JsonNode object, java.util.Set allowed) {
+ if (!object.isObject()) throw invalid("noncanonical protocol object");
+ object.fieldNames().forEachRemaining(key -> {
+ if (!allowed.contains(key)) throw invalid("noncanonical protocol control field");
+ });
+ }
+
+ private T savepoint(String name, java.util.function.Supplier action) {
+ sql.execute("SAVEPOINT " + name);
+ try {
+ T value = action.get();
+ sql.execute("RELEASE SAVEPOINT " + name);
+ return value;
+ } catch (RuntimeException | Error failure) {
+ try {
+ sql.execute("ROLLBACK TO SAVEPOINT " + name);
+ sql.execute("RELEASE SAVEPOINT " + name);
+ } catch (RuntimeException rollbackFailure) {
+ // A failed savepoint rollback must not leave partial rows committable.
+ try { sql.execute("ROLLBACK"); } catch (RuntimeException ignored) { }
+ throw new CommittedEventException("ROLLBACK_FAILED", "committed-event rollback failed");
+ }
+ throw failure;
+ }
+ }
+
+ private void requireTransaction() {
+ if (!sql.hasActiveTransaction()) throw invalid("an active SQL transaction is required");
+ }
+
+ private static long number(Object value) { return ((Number) value).longValue(); }
+ private static boolean blank(String value) { return value == null || value.isBlank(); }
+ private static CommittedEventException invalid(String reason) {
+ return new CommittedEventException("INVALID_ENVELOPE", "invalid committed-event envelope: " + reason);
+ }
+ private static String digest(String payload) {
+ try {
+ byte[] hash = MessageDigest.getInstance("SHA-256")
+ .digest(payload.getBytes(StandardCharsets.UTF_8));
+ return HexFormat.of().formatHex(hash);
+ } catch (NoSuchAlgorithmException impossible) {
+ throw new AssertionError("JDK SHA-256 is unavailable", impossible);
+ }
+ }
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedMutationRecorder.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedMutationRecorder.java
new file mode 100644
index 00000000..1e45db2f
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedMutationRecorder.java
@@ -0,0 +1,111 @@
+package io.teaql.core.sqlite;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.teaql.core.GeneratedSchemaBootstrap;
+import io.teaql.core.CommittedEventSchemaContributor;
+import io.teaql.core.MutationIntent;
+import io.teaql.core.SchemaExecutor;
+import io.teaql.core.TraceNode;
+import io.teaql.core.UserContext;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+import io.teaql.runtime.CommittedMutationFact;
+import io.teaql.runtime.CommittedMutationRecorder;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/** SQLite source adapter that records one graph-save result inside its write transaction. */
+public final class SqliteCommittedMutationRecorder
+ implements CommittedMutationRecorder, CommittedEventSchemaContributor {
+ private static final ObjectMapper JSON = new ObjectMapper();
+ private final SqlExecutionAdapter adapter;
+ private final SqliteCommittedEventStore events;
+ private final String sourceRuntimeId;
+ private final String modelHash;
+
+ public SqliteCommittedMutationRecorder(SqlExecutionAdapter adapter,
+ String sourceRuntimeId, String modelHash) {
+ this.adapter = Objects.requireNonNull(adapter, "adapter");
+ this.events = new SqliteCommittedEventStore(adapter);
+ if (sourceRuntimeId == null || sourceRuntimeId.isBlank()
+ || modelHash == null || !modelHash.matches("sha256:[0-9a-f]{64}")) {
+ throw new IllegalArgumentException("A source runtime ID and accepted SHA-256 model hash are required");
+ }
+ this.sourceRuntimeId = sourceRuntimeId;
+ this.modelHash = modelHash;
+ }
+
+ @Override
+ public void ensureSchema(UserContext context, SchemaExecutor.Invocation invocation) {
+ SchemaExecutor.Invocation.requireContextOwned(invocation);
+ Objects.requireNonNull(context, "context");
+ events.ensureSchema();
+ }
+ public SqliteCommittedEventStore eventStore() { return events; }
+
+ @Override
+ public void record(UserContext context, Object transactionResource,
+ MutationIntent intent, List facts) {
+ Objects.requireNonNull(context, "context");
+ Objects.requireNonNull(intent, "intent");
+ if (transactionResource != adapter || !adapter.hasActiveTransaction()) {
+ throw new IllegalStateException("Committed graph capture requires the source provider transaction");
+ }
+ if (facts == null || facts.isEmpty()) {
+ throw new IllegalArgumentException("Committed graph capture requires at least one fact");
+ }
+ String changeSetId = UUID.randomUUID().toString();
+ String committedAt = Instant.now().toString();
+ String actor = context.getAttribute(GeneratedSchemaBootstrap.AUDIT_ACTOR_ATTRIBUTE, String.class);
+ events.append(sourceRuntimeId, changeSetId, modelHash, sequence -> {
+ Map envelope = new LinkedHashMap<>();
+ envelope.put("protocol_version", 1);
+ envelope.put("change_set_id", changeSetId);
+ envelope.put("source_runtime_id", sourceRuntimeId);
+ envelope.put("source_sequence", Long.toString(sequence));
+ envelope.put("source_model_hash", modelHash);
+ envelope.put("source_transaction_id", changeSetId);
+ envelope.put("committed_at", committedAt);
+ Map audit = new LinkedHashMap<>();
+ audit.put("actor_ref", actor);
+ audit.put("comment", intent.comment());
+ // MutationIntent currently has no separate business purpose field.
+ audit.put("purpose", "replicate committed graph");
+ envelope.put("audit", audit);
+ List> mutations = new ArrayList<>(facts.size());
+ for (CommittedMutationFact fact : facts) {
+ Map mutation = new LinkedHashMap<>();
+ mutation.put("entity_type", fact.entityType());
+ mutation.put("entity_id", fact.entityId());
+ mutation.put("operation", fact.operation());
+ mutation.put("base_version", fact.baseVersion());
+ mutation.put("committed_version", fact.committedVersion());
+ mutation.put("fields", fact.fields());
+ List> chain = new ArrayList<>();
+ for (TraceNode node : fact.traceChain()) {
+ Map step = new LinkedHashMap<>();
+ step.put("kind", node.getKind().name());
+ step.put("name", Objects.toString(node.getName(), ""));
+ step.put("entity_id", node.getEntityId() == null ? null : node.getEntityId().toString());
+ step.put("comment", Objects.toString(node.getComment(), ""));
+ chain.add(step);
+ }
+ mutation.put("trace_chain", chain);
+ mutations.add(mutation);
+ }
+ envelope.put("mutations", mutations);
+ try {
+ return JSON.writeValueAsString(envelope);
+ } catch (JsonProcessingException failure) {
+ // Do not attach values or serializer causes to public errors.
+ throw new IllegalStateException("Committed mutation payload encoding failed");
+ }
+ });
+ }
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedObservation.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedObservation.java
new file mode 100644
index 00000000..95e066f5
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedObservation.java
@@ -0,0 +1,115 @@
+package io.teaql.core.sqlite;
+
+import io.teaql.core.UserContext;
+import io.teaql.core.meta.EntityMetaFactory;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+import io.teaql.runtime.RuntimeTelemetry;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+/** Durable infrastructure observations. Source ACK is never labeled target watermark. */
+public final class SqliteCommittedObservation {
+ public record SourceObservation(long sourceHighWatermark,long ackedSequence,long sequenceLag,
+ long backlog,long pendingSnapshots,long attempts,long retries,long failures,
+ long rejections,long deadLetters,long unresolvedAttempts) { }
+ public record TargetObservation(long checkpoint,long snapshotBaseline,long inboxRows,
+ OptionalLong sequenceLag,long rejections) { }
+ public record DeadLetter(String kind,long sequence,String identityFingerprint,
+ String payloadFingerprint,long attempts,String reason) { }
+ record Attempt(String kind,long sequence,long ordinal) { }
+ private final SqlExecutionAdapter sql;
+ private final String source,target,model;
+ private static final Set PERMANENT=Set.of("UNAUTHORIZED_SOURCE","UNAUTHORIZED_TARGET",
+ "UNKNOWN_KEY","INVALID_SIGNATURE","INVALID_ENVELOPE","MODEL_MISMATCH",
+ "SEQUENCE_GAP","DUPLICATE_CONFLICT","VERSION_CONFLICT","SEQUENCE_EXHAUSTED",
+ "INVALID_SNAPSHOT","SNAPSHOT_CONFLICT","TARGET_NOT_EMPTY");
+ public SqliteCommittedObservation(SqlExecutionAdapter sql,String source,String target,String model){
+ this.sql=Objects.requireNonNull(sql,"sql");
+ if(source==null||source.isBlank()||target==null||target.isBlank()||model==null||!model.matches("sha256:[0-9a-f]{64}"))throw unavailable();
+ this.source=source;this.target=target;this.model=model;
+ }
+ static void ensureSchema(SqlExecutionAdapter sql){
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_delivery_journal (source_runtime_id TEXT NOT NULL,target_id TEXT NOT NULL,kind TEXT NOT NULL CHECK(kind IN ('event','snapshot')),source_sequence INTEGER NOT NULL CHECK(source_sequence>=0),model_hash TEXT NOT NULL,identity_fingerprint TEXT NOT NULL,payload_fingerprint TEXT NOT NULL,attempts INTEGER NOT NULL CHECK(typeof(attempts)='integer' AND attempts>0),failures INTEGER NOT NULL CHECK(typeof(failures)='integer' AND failures>=0),rejections INTEGER NOT NULL CHECK(typeof(rejections)='integer' AND rejections>=0),dead_letter INTEGER NOT NULL CHECK(dead_letter IN (0,1)),last_outcome TEXT NOT NULL CHECK(last_outcome IN ('STARTED','ACKNOWLEDGED','FAILED')),last_reason TEXT NOT NULL,dead_reason TEXT NOT NULL,PRIMARY KEY(source_runtime_id,target_id,kind,source_sequence))");
+ }
+ Attempt begin(String kind,long sequence,String identity,String payload){
+ if(!Set.of("event","snapshot").contains(kind)||sequence<0||identity==null||payload==null||!payload.matches("[0-9a-f]{64}"))throw unavailable();
+ return transaction(()->{
+ var rows=sql.queryForList("SELECT * FROM teaql_event_delivery_journal WHERE source_runtime_id=? AND target_id=? AND kind=? AND source_sequence=?",new Object[]{source,target,kind,sequence});
+ long ordinal=1;
+ if(rows.isEmpty())sql.update("INSERT INTO teaql_event_delivery_journal VALUES(?,?,?,?,?,?,?,1,0,0,0,'STARTED','NONE','NONE')",new Object[]{source,target,kind,sequence,model,fingerprint(identity),payload});
+ else{
+ var row=rows.get(0);long prior=number(row.get("attempts"));
+ if(!model.equals(row.get("model_hash"))||!fingerprint(identity).equals(row.get("identity_fingerprint"))||!payload.equals(row.get("payload_fingerprint"))||"ACKNOWLEDGED".equals(row.get("last_outcome")))throw conflict();
+ ordinal=Math.addExact(prior,1);
+ if(sql.update("UPDATE teaql_event_delivery_journal SET attempts=?,last_outcome='STARTED',last_reason='NONE' WHERE source_runtime_id=? AND target_id=? AND kind=? AND source_sequence=? AND attempts=?",new Object[]{ordinal,source,target,kind,sequence,prior})!=1)throw conflict();
+ }
+ return new Attempt(kind,sequence,ordinal);
+ });
+ }
+ void failed(Attempt attempt,RuntimeException error){
+ String code=error instanceof SqliteCommittedReceiver.ReceiverException r?r.code():error instanceof SqliteCommittedEventStore.CommittedEventException e?e.code():"TRANSPORT_UNAVAILABLE";
+ String normalized=code.startsWith("SNAPSHOT_")&&PERMANENT.contains(code.substring(9))?code.substring(9):code;
+ boolean permanent=PERMANENT.contains(normalized);
+ String safe=permanent?normalized:Set.of("ACK_MISMATCH","AUDIT_UNAVAILABLE","CURSOR_CONFLICT").contains(code)?code:"TRANSPORT_UNAVAILABLE";
+ transaction(()->{if(sql.update("UPDATE teaql_event_delivery_journal SET failures=failures+1,rejections=rejections+?,dead_letter=CASE WHEN ?=1 THEN 1 ELSE dead_letter END,dead_reason=CASE WHEN ?=1 THEN ? ELSE dead_reason END,last_outcome='FAILED',last_reason=? WHERE source_runtime_id=? AND target_id=? AND kind=? AND source_sequence=? AND attempts=? AND last_outcome='STARTED'",new Object[]{permanent?1:0,permanent?1:0,permanent?1:0,safe,safe,source,target,attempt.kind(),attempt.sequence(),attempt.ordinal()})!=1)throw conflict();return null;});
+ }
+ T acknowledged(Attempt attempt,Supplier cursorUpdate){
+ return transaction(()->{
+ T value=cursorUpdate.get();
+ if(sql.update("UPDATE teaql_event_delivery_journal SET last_outcome='ACKNOWLEDGED',last_reason='NONE',dead_letter=0,dead_reason='NONE' WHERE source_runtime_id=? AND target_id=? AND kind=? AND source_sequence=? AND attempts=? AND last_outcome='STARTED'",new Object[]{source,target,attempt.kind(),attempt.sequence(),attempt.ordinal()})!=1)throw conflict();
+ return value;
+ });
+ }
+ public SourceObservation source(UserContext context){
+ require(context);
+ SourceObservation value=transaction(()->{
+ if(count("SELECT count(*) n FROM teaql_event_delivery_journal WHERE source_runtime_id=? AND target_id=? AND model_hash<>?",new Object[]{source,target,model})!=0)throw unavailable();
+ var cursor=sql.queryForList("SELECT acked_sequence,model_hash FROM teaql_event_delivery_cursor WHERE source_runtime_id=? AND target_id=?",new Object[]{source,target});
+ long ack=cursor.isEmpty()?0:number(cursor.get(0).get("acked_sequence"));
+ if(!cursor.isEmpty()&&!model.equals(cursor.get(0).get("model_hash")))throw unavailable();
+ var highRows=sql.queryForList("SELECT current_sequence FROM teaql_event_sequence WHERE source_runtime_id=?",new Object[]{source});long high=highRows.isEmpty()?0:number(highRows.get(0).get("current_sequence"));
+ if(high?",new Object[]{source,ack});
+ long pending=count("SELECT count(*) n FROM teaql_event_snapshot_publication WHERE source_runtime_id=? AND target_id=? AND model_hash=? AND acknowledged=0",new Object[]{source,target,model});
+ return new SourceObservation(high,ack,high-ack,backlog,pending,number(sums.get("attempts")),number(sums.get("retries")),number(sums.get("failures")),number(sums.get("rejections")),number(sums.get("dead")),number(sums.get("unresolved")));
+ });
+ emit(context,"source",Map.of("source_watermark",value.sourceHighWatermark(),"source_ack",value.ackedSequence(),"sequence_lag",value.sequenceLag(),"backlog",value.backlog(),"pending_snapshots",value.pendingSnapshots(),"attempts",value.attempts(),"retries",value.retries(),"failures",value.failures(),"rejections",value.rejections(),"dead_letters",value.deadLetters()),Map.of("unresolved_attempts",value.unresolvedAttempts()));
+ return value;
+ }
+ public TargetObservation target(UserContext context,OptionalLong observedSourceHigh){
+ require(context);Objects.requireNonNull(observedSourceHigh,"observedSourceHigh");
+ TargetObservation value=transaction(()->{
+ var rows=sql.queryForList("SELECT source_sequence,model_hash FROM teaql_event_checkpoint WHERE source_runtime_id=?",new Object[]{source});
+ long checkpoint=rows.isEmpty()?0:number(rows.get(0).get("source_sequence"));
+ if(!rows.isEmpty()&&!model.equals(rows.get(0).get("model_hash")))throw unavailable();
+ if(observedSourceHigh.isPresent()&&observedSourceHigh.getAsLong()checkpoint)throw unavailable();
+ return new TargetObservation(checkpoint,n,count("SELECT count(*) n FROM teaql_event_inbox WHERE source_runtime_id=?",new Object[]{source}),observedSourceHigh.isPresent()?OptionalLong.of(observedSourceHigh.getAsLong()-checkpoint):OptionalLong.empty(),count("SELECT count(*) n FROM teaql_event_rejection WHERE target_fingerprint=?",new Object[]{fingerprint(target)}));
+ });
+ var attrs=new LinkedHashMap();attrs.put("target_watermark",value.checkpoint());attrs.put("snapshot_baseline",value.snapshotBaseline());attrs.put("inbox_rows",value.inboxRows());attrs.put("rejections",value.rejections());if(value.sequenceLag().isPresent())attrs.put("sequence_lag",value.sequenceLag().getAsLong());emit(context,"target",attrs,Map.of());return value;
+ }
+ public List deadLetters(UserContext context,int limit){
+ require(context);if(limit<1||limit>1000)throw unavailable();
+ return transaction(()->sql.queryForList("SELECT kind,source_sequence,identity_fingerprint,payload_fingerprint,attempts,dead_reason FROM teaql_event_delivery_journal WHERE source_runtime_id=? AND target_id=? AND model_hash=? AND dead_letter=1 ORDER BY source_sequence,kind LIMIT ?",new Object[]{source,target,model,limit}).stream().map(r->new DeadLetter((String)r.get("kind"),number(r.get("source_sequence")),(String)r.get("identity_fingerprint"),(String)r.get("payload_fingerprint"),number(r.get("attempts")),(String)r.get("dead_reason"))).toList());
+ }
+ private void emit(UserContext context,String name,Map values,Map extra){
+ var attributes=new LinkedHashMap();attributes.put("teaql.replication.source_fingerprint",fingerprint(source));attributes.put("teaql.replication.target_fingerprint",fingerprint(target));values.forEach((k,v)->attributes.put("teaql.replication."+k,v));extra.forEach((k,v)->attributes.put("teaql.replication."+k,v));
+ RuntimeTelemetry.startSafely(context.capability(RuntimeTelemetry.class),new RuntimeTelemetry.Operation("replication",name,attributes)).success();
+ }
+ private long count(String query,Object[] args){return number(sql.queryForList(query,args).get(0).get("n"));}
+ private void require(UserContext context){Objects.requireNonNull(context,"context");EntityMetaFactory.requireFrom(context);if(sql.hasActiveTransaction())throw unavailable();}
+ private T transaction(Supplier action){
+ if(sql.hasActiveTransaction())throw unavailable();
+ var value=new AtomicReference();sql.executeInTransaction(()->{if(!sql.hasActiveTransaction())throw unavailable();value.set(action.get());});return value.get();
+ }
+ private static long number(Object value){if(!(value instanceof Number n)||n instanceof Double||n instanceof Float||n.longValue()<0)throw unavailable();return n.longValue();}
+ static String fingerprint(String value){try{return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(value.getBytes(StandardCharsets.UTF_8)));}catch(java.security.NoSuchAlgorithmException impossible){throw new IllegalStateException(impossible);}}
+ private static SqliteCommittedEventStore.CommittedEventException unavailable(){return new SqliteCommittedEventStore.CommittedEventException("OBSERVATION_UNAVAILABLE","committed delivery observation is unavailable");}
+ private static SqliteCommittedEventStore.CommittedEventException conflict(){return new SqliteCommittedEventStore.CommittedEventException("CURSOR_CONFLICT","committed delivery attempt changed concurrently");}
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedOutboxPublisher.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedOutboxPublisher.java
new file mode 100644
index 00000000..9607ca6f
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedOutboxPublisher.java
@@ -0,0 +1,143 @@
+package io.teaql.core.sqlite;
+
+import io.teaql.core.sqlite.SqliteCommittedEventStore.CommittedEventException;
+import io.teaql.core.sqlite.SqliteCommittedEventStore.InboxOutcome;
+import io.teaql.core.sqlite.SqliteCommittedEventStore.OutboxRecord;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Bounded ordered publication from one durable SQLite Outbox stream. The
+ * supplied sink must own its authentication and network retry budget; this
+ * publisher never treats a callback as proof of remote identity.
+ */
+public final class SqliteCommittedOutboxPublisher {
+ @FunctionalInterface
+ public interface DeliverySink {
+ InboxOutcome deliver(OutboxRecord record);
+ }
+
+ public record DeliveryStatus(long ackedSequence, long sourceHighWatermark, long backlog) { }
+ public record BatchOutcome(int acknowledged, DeliveryStatus status) { }
+
+ private final SqlExecutionAdapter sql;
+ private final SqliteCommittedEventStore events;
+ private final String sourceRuntimeId;
+ private final String targetId;
+ private final String modelHash;
+
+ public SqliteCommittedOutboxPublisher(SqlExecutionAdapter sql, String sourceRuntimeId,
+ String targetId, String modelHash) {
+ this.sql = Objects.requireNonNull(sql, "sql");
+ if (sourceRuntimeId == null || sourceRuntimeId.isBlank()
+ || targetId == null || targetId.isBlank()
+ || modelHash == null || !modelHash.matches("sha256:[0-9a-f]{64}")) {
+ throw failure("INVALID_ENVELOPE", "delivery source, target or model identity is invalid");
+ }
+ this.sourceRuntimeId = sourceRuntimeId;
+ this.targetId = targetId;
+ this.modelHash = modelHash;
+ this.events = new SqliteCommittedEventStore(sql);
+ }
+
+ public DeliveryStatus status() {
+ List> rows = sql.queryForList(
+ "SELECT acked_sequence, model_hash FROM teaql_event_delivery_cursor "
+ + "WHERE source_runtime_id = ? AND target_id = ?",
+ new Object[] {sourceRuntimeId, targetId});
+ long acked = 0;
+ if (!rows.isEmpty()) {
+ if (!modelHash.equals(rows.get(0).get("model_hash"))) {
+ throw failure("MODEL_MISMATCH", "delivery cursor model hash mismatch");
+ }
+ acked = number(rows.get(0).get("acked_sequence"));
+ }
+ List> sourceRows = sql.queryForList(
+ "SELECT current_sequence FROM teaql_event_sequence WHERE source_runtime_id = ?",
+ new Object[] {sourceRuntimeId});
+ long highWatermark = sourceRows.isEmpty() ? 0
+ : number(sourceRows.get(0).get("current_sequence"));
+ if (highWatermark < acked) {
+ throw failure("INVALID_ENVELOPE", "delivery cursor is ahead of source sequence");
+ }
+ return new DeliveryStatus(acked, highWatermark, highWatermark - acked);
+ }
+
+ /**
+ * One attempt per event, at most {@code limit} events. The source cursor
+ * advances only after an acknowledgement for the exact sequence. If the
+ * target applied an event but the response is lost, a later call/restart
+ * re-delivers it and the target Inbox supplies a duplicate acknowledgement.
+ */
+ public BatchOutcome publishBatch(int limit, DeliverySink sink) {
+ Objects.requireNonNull(sink, "sink");
+ if (limit < 1 || limit > 1000) throw failure("INVALID_ENVELOPE", "invalid delivery read limit");
+ claimCursor();
+ DeliveryStatus before = status();
+ List records = events.outboxAfter(sourceRuntimeId, before.ackedSequence(), limit);
+ if (records.isEmpty() && before.backlog() > 0) {
+ throw failure("SEQUENCE_GAP", "source Outbox has an unacknowledged gap");
+ }
+ long acked = before.ackedSequence();
+ int count = 0;
+ for (OutboxRecord record : records) {
+ if (acked == Long.MAX_VALUE) {
+ throw failure("SEQUENCE_EXHAUSTED", "source sequence exhausted");
+ }
+ if (record.sourceSequence() != acked + 1) {
+ throw failure("SEQUENCE_GAP", "source Outbox sequence is not contiguous");
+ }
+ if (!modelHash.equals(record.modelHash())) {
+ throw failure("MODEL_MISMATCH", "source Outbox model hash mismatch");
+ }
+ var journal = new SqliteCommittedObservation(sql,sourceRuntimeId,targetId,modelHash);
+ var attempt = journal.begin("event",record.sourceSequence(),record.changeSetId(),record.payloadSha256());
+ try {
+ InboxOutcome outcome = sink.deliver(record);
+ if (outcome == null || outcome.checkpoint() != record.sourceSequence())
+ throw failure("ACK_MISMATCH", "target acknowledgement does not match delivered sequence");
+ } catch (RuntimeException error) { journal.failed(attempt,error); throw error; }
+ long prior = acked;
+ journal.acknowledged(attempt, () -> {
+ int changed = sql.update("UPDATE teaql_event_delivery_cursor SET acked_sequence=? "
+ + "WHERE source_runtime_id=? AND target_id=? AND acked_sequence=? AND model_hash=?",
+ new Object[] {record.sourceSequence(),sourceRuntimeId,targetId,prior,modelHash});
+ if (changed != 1) {
+ throw failure("CURSOR_CONFLICT", "delivery cursor changed concurrently");
+ }
+ return null;
+ });
+ acked = record.sourceSequence();
+ count++;
+ }
+ return new BatchOutcome(count, status());
+ }
+
+ private void claimCursor() {
+ if (sql.hasActiveTransaction()) throw failure("INVALID_ENVELOPE", "publisher requires a standalone transaction");
+ sql.executeInTransaction(() -> {
+ if (!sql.hasActiveTransaction()) throw failure("INVALID_ENVELOPE", "publisher adapter must provide an active SQL transaction");
+ var prepared = sql.queryForList("SELECT source_sequence,model_hash,acknowledged FROM teaql_event_snapshot_publication WHERE source_runtime_id=? AND target_id=?",
+ new Object[]{sourceRuntimeId,targetId});
+ if (!prepared.isEmpty()) {
+ var row = prepared.get(0);
+ if (number(row.get("acknowledged")) == 0) throw failure("SNAPSHOT_PENDING", "snapshot bootstrap must be acknowledged before incremental delivery");
+ var cursor = sql.queryForList("SELECT acked_sequence,model_hash FROM teaql_event_delivery_cursor WHERE source_runtime_id=? AND target_id=?",new Object[]{sourceRuntimeId,targetId});
+ if (number(row.get("acknowledged")) != 1 || !modelHash.equals(row.get("model_hash")) || cursor.size()!=1
+ || !modelHash.equals(cursor.get(0).get("model_hash")) || number(cursor.get(0).get("acked_sequence")) REJECTIONS=Set.of("INVALID_ENVELOPE","MODEL_MISMATCH",
+ "SEQUENCE_GAP","DUPLICATE_CONFLICT","VERSION_CONFLICT","SEQUENCE_EXHAUSTED");
+ private final SqlExecutionAdapter sql;
+ private final String source,target,model;
+ private final Map keys;
+ private final SqliteModelCheckedCommittedTarget delegate;
+ private final SqliteCommittedSnapshot snapshots;
+ private final SqliteCommittedSnapshot.Limits snapshotLimits;
+ private final SqliteCommittedSnapshot.ImportMode snapshotMode;
+
+ public record Delivery(OutboxRecord event,String keyId,String targetId,byte[] signature) {
+ public Delivery {signature=signature==null?new byte[0]:signature.clone();}
+ @Override public byte[] signature(){return signature.clone();}
+ @Override public String toString(){return "CommittedDelivery[content=]";}
+ }
+ public record SnapshotDelivery(SqliteCommittedSnapshot.SnapshotRecord snapshot,
+ String keyId,String targetId,byte[] signature) {
+ public SnapshotDelivery {signature=signature==null?new byte[0]:signature.clone();}
+ @Override public byte[] signature(){return signature.clone();}
+ @Override public String toString(){return "CommittedSnapshotDelivery[content=]";}
+ }
+ public static final class ReceiverException extends RuntimeException {
+ private final String code;
+ private ReceiverException(String code){super("committed receiver: "+code);this.code=code;}
+ public String code(){return code;}
+ }
+ public SqliteCommittedReceiver(SqlExecutionAdapter sql,String source,String target,
+ String model,Map keys) {
+ this(sql,source,target,model,keys,SqliteCommittedSnapshot.Limits.defaults());
+ }
+ /** Limits are trusted startup policy, never caller-controlled wire options. */
+ public SqliteCommittedReceiver(SqlExecutionAdapter sql,String source,String target,
+ String model,Map keys,SqliteCommittedSnapshot.Limits snapshotLimits) {
+ this(sql,source,target,model,keys,snapshotLimits,SqliteCommittedSnapshot.ImportMode.REQUIRE_EMPTY);
+ }
+ /** Explicit local policy for bootstrap-populated targets, never a wire flag. */
+ public SqliteCommittedReceiver(SqlExecutionAdapter sql,String source,String target,
+ String model,Map keys,SqliteCommittedSnapshot.Limits snapshotLimits,
+ SqliteCommittedSnapshot.ImportMode snapshotMode) {
+ this.sql=Objects.requireNonNull(sql,"sql");
+ if(snapshotLimits==null||snapshotLimits.maxPayloadBytes()>MAX_SNAPSHOT_PAYLOAD||snapshotMode==null)
+ throw new IllegalArgumentException("invalid receiver snapshot limits");
+ this.snapshotLimits=snapshotLimits;
+ this.snapshotMode=snapshotMode;
+ try {
+ field(source,MAX_ID);field(target,MAX_ID);
+ if(model==null||!model.matches("sha256:[0-9a-f]{64}")||keys==null||keys.isEmpty()||keys.size()>16)throw rejected("INVALID_ENVELOPE");
+ Map copy=new HashMap<>();
+ for(var entry:keys.entrySet()){field(entry.getKey(),MAX_ID);key(entry.getValue());copy.put(entry.getKey(),entry.getValue().clone());}
+ this.keys=Map.copyOf(copy);
+ }catch(ReceiverException failure){throw new IllegalArgumentException("invalid committed receiver configuration");}
+ this.source=source;this.target=target;this.model=model;
+ this.delegate=new SqliteModelCheckedCommittedTarget(sql,source,model);
+ this.snapshots=new SqliteCommittedSnapshot(sql,source,model);
+ }
+ /** Explicit provider-infrastructure installation; no domain-row seeding. */
+ public void ensureSchema(UserContext context) {
+ EntityMetaFactory.requireFrom(Objects.requireNonNull(context,"context"));
+ if(sql.hasActiveTransaction())throw rejected("AMBIENT_TRANSACTION");
+ delegate.ensureSchema();
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_rejection (audit_id INTEGER PRIMARY KEY, "
+ +"reason_code TEXT NOT NULL, source_sequence INTEGER NOT NULL, source_fingerprint TEXT NOT NULL, "
+ +"change_fingerprint TEXT NOT NULL,target_fingerprint TEXT NOT NULL,payload_fingerprint TEXT NOT NULL)");
+ }
+ /** Original model facts remain unchanged; authenticated duplicate ACKs stay stable. */
+ public SqliteCommittedEventStore.InboxOutcome receive(UserContext context,Delivery delivery) {
+ Objects.requireNonNull(context,"context");
+ // Infrastructure wiring error: never join/commit a caller's transaction.
+ if(sql.hasActiveTransaction())throw rejected("AMBIENT_TRANSACTION");
+ try {
+ validate(delivery);
+ OutboxRecord event=delivery.event();
+ if(!source.equals(event.sourceRuntimeId()))throw rejected("UNAUTHORIZED_SOURCE");
+ if(!target.equals(delivery.targetId()))throw rejected("UNAUTHORIZED_TARGET");
+ if(!model.equals(event.modelHash()))throw rejected("MODEL_MISMATCH");
+ byte[] secret=keys.get(delivery.keyId());
+ if(secret==null)throw rejected("UNKNOWN_KEY");
+ if(!MessageDigest.isEqual(tag(delivery.keyId(),delivery.targetId(),secret,event),delivery.signature))throw rejected("INVALID_SIGNATURE");
+ return delegate.apply(context,event);
+ }catch(ReceiverException failure){
+ audit(delivery,failure.code());throw failure;
+ }catch(SqliteCommittedEventStore.CommittedEventException failure){
+ // Unknown/failed transaction outcomes are not security rejections.
+ if(!REJECTIONS.contains(failure.code()))throw failure;
+ audit(delivery,failure.code());throw rejected(failure.code());
+ }
+ }
+ /** Source-side signing of original bytes; requires a deployment-owned secret. */
+ public static Delivery sign(String keyId,String target,byte[] secret,OutboxRecord event) {
+ field(keyId,MAX_ID);field(target,MAX_ID);key(secret);record(event);
+ return new Delivery(event,keyId,target,tag(keyId,target,secret,event));
+ }
+ /** Separate type domain; zero is a valid snapshot baseline, never an event sequence. */
+ public static SnapshotDelivery signSnapshot(String keyId,String target,byte[] secret,
+ SqliteCommittedSnapshot.SnapshotRecord snapshot) {
+ field(keyId,MAX_ID);field(target,MAX_ID);key(secret);snapshot(snapshot,MAX_SNAPSHOT_PAYLOAD);
+ return new SnapshotDelivery(snapshot,keyId,target,snapshotTag(keyId,target,secret,snapshot));
+ }
+ /** Authenticate before restore. Deliveries cannot select or relax startup restore policy. */
+ public SqliteCommittedSnapshotPublisher.Receipt receiveSnapshot(UserContext context,SnapshotDelivery delivery) {
+ Objects.requireNonNull(context,"context");
+ if(sql.hasActiveTransaction())throw rejected("AMBIENT_TRANSACTION");
+ try {
+ if(delivery==null)throw rejected("INVALID_ENVELOPE");
+ field(delivery.keyId(),MAX_ID);field(delivery.targetId(),MAX_ID);
+ snapshot(delivery.snapshot(),snapshotLimits.maxPayloadBytes());
+ if(delivery.signature.length!=32)throw rejected("INVALID_ENVELOPE");
+ var record=delivery.snapshot();
+ if(!source.equals(record.sourceRuntimeId()))throw rejected("UNAUTHORIZED_SOURCE");
+ if(!target.equals(delivery.targetId()))throw rejected("UNAUTHORIZED_TARGET");
+ if(!model.equals(record.modelHash()))throw rejected("MODEL_MISMATCH");
+ byte[] secret=keys.get(delivery.keyId());if(secret==null)throw rejected("UNKNOWN_KEY");
+ if(!MessageDigest.isEqual(snapshotTag(delivery.keyId(),delivery.targetId(),secret,record),delivery.signature))throw rejected("INVALID_SIGNATURE");
+ return SqliteCommittedSnapshotPublisher.initializeAndAcknowledge(context,snapshots,record,target,snapshotMode,snapshotLimits);
+ }catch(ReceiverException failure){audit(delivery,failure.code());throw failure;}
+ catch(SqliteCommittedEventStore.CommittedEventException failure){
+ String code="INVALID_ENVELOPE".equals(failure.code())?"INVALID_SNAPSHOT":failure.code();
+ if(!REJECTIONS.contains(code)&&!Set.of("INVALID_SNAPSHOT","SNAPSHOT_CONFLICT").contains(code))throw failure;
+ audit(delivery,code);throw rejected(code);
+ }
+ }
+ private static void validate(Delivery delivery){
+ if(delivery==null)throw rejected("INVALID_ENVELOPE");
+ field(delivery.keyId(),MAX_ID);field(delivery.targetId(),MAX_ID);record(delivery.event());
+ if(delivery.signature.length!=32)throw rejected("INVALID_ENVELOPE");
+ }
+ private static void record(OutboxRecord e){
+ if(e==null||e.sourceSequence()<=0)throw rejected("INVALID_ENVELOPE");
+ field(e.sourceRuntimeId(),MAX_ID);field(e.changeSetId(),MAX_ID);field(e.modelHash(),MAX_ID);field(e.payloadSha256(),MAX_ID);field(e.payloadJson(),MAX_PAYLOAD);
+ }
+ private static void snapshot(SqliteCommittedSnapshot.SnapshotRecord s,int byteLimit){
+ if(s==null||s.sourceSequence()<0)throw rejected("INVALID_ENVELOPE");
+ field(s.sourceRuntimeId(),MAX_ID);field(s.snapshotId(),MAX_ID);field(s.modelHash(),MAX_ID);field(s.payloadSha256(),MAX_ID);field(s.payloadJson(),byteLimit);
+ }
+ private static byte[] field(String text,int limit){
+ if(text==null||text.isEmpty()||text.length()>limit||!StandardCharsets.UTF_8.newEncoder().canEncode(text))throw rejected("INVALID_ENVELOPE");
+ byte[] bytes=text.getBytes(StandardCharsets.UTF_8);if(bytes.length>limit)throw rejected("INVALID_ENVELOPE");return bytes;
+ }
+ private static void key(byte[] secret){if(secret==null||secret.length<32||secret.length>128)throw rejected("INVALID_ENVELOPE");}
+ private static byte[] tag(String keyId,String target,byte[] secret,OutboxRecord event){
+ return hmac(secret,"teaql.committed.receiver.v1\0",keyId,target,event.sourceRuntimeId(),Long.toString(event.sourceSequence()),event.changeSetId(),event.modelHash(),event.payloadSha256(),event.payloadJson());
+ }
+ private static byte[] snapshotTag(String keyId,String target,byte[] secret,SqliteCommittedSnapshot.SnapshotRecord s){
+ return hmac(secret,"teaql.committed.snapshot.receiver.v1\0",keyId,target,s.sourceRuntimeId(),Long.toString(s.sourceSequence()),s.snapshotId(),s.modelHash(),s.payloadSha256(),s.payloadJson());
+ }
+ private static byte[] hmac(byte[] secret,String domain,String... fields){
+ try {
+ Mac mac=Mac.getInstance("HmacSHA256");mac.init(new SecretKeySpec(secret,"HmacSHA256"));
+ mac.update(domain.getBytes(StandardCharsets.UTF_8));
+ for(String value:fields){byte[] bytes=value.getBytes(StandardCharsets.UTF_8);mac.update(ByteBuffer.allocate(4).putInt(bytes.length).array());mac.update(bytes);}
+ return mac.doFinal();
+ }catch(GeneralSecurityException failure){throw new IllegalStateException("standard HMAC-SHA256 unavailable");}
+ }
+ private void audit(Delivery delivery,String code){
+ OutboxRecord e=delivery==null?null:delivery.event();
+ audit(code,e==null?0:e.sourceSequence(),e==null?null:e.sourceRuntimeId(),e==null?null:e.changeSetId(),e==null?null:e.payloadJson(),MAX_PAYLOAD);
+ }
+ private void audit(SnapshotDelivery delivery,String code){
+ var s=delivery==null?null:delivery.snapshot();
+ audit("SNAPSHOT_"+code,s==null?0:s.sourceSequence(),s==null?null:s.sourceRuntimeId(),s==null?null:s.snapshotId(),s==null?null:s.payloadJson(),MAX_SNAPSHOT_PAYLOAD);
+ }
+ private void audit(String code,long sequence,String sourceId,String identity,String payload,int maxPayload){
+ try {
+ if(sql.hasActiveTransaction())throw new IllegalStateException("audit transaction unavailable");
+ Object[] values={code,Math.max(0,sequence),fingerprint(sourceId,MAX_ID),fingerprint(identity,MAX_ID),fingerprint(target,MAX_ID),fingerprint(payload,maxPayload)};
+ sql.executeInTransaction(()->{if(sql.update("INSERT INTO teaql_event_rejection(reason_code,source_sequence,source_fingerprint,change_fingerprint,target_fingerprint,payload_fingerprint) VALUES (?,?,?,?,?,?)",values)!=1)throw new IllegalStateException("audit row missing");});
+ }catch(RuntimeException failure){throw rejected("AUDIT_UNAVAILABLE");}
+ }
+ private static String fingerprint(String text,int limit){
+ if(text==null)return "missing";
+ byte[] bytes;try{bytes=field(text,limit);}catch(ReceiverException invalid){return "invalid_or_oversize";}
+ try{return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes));}
+ catch(GeneralSecurityException impossible){throw new IllegalStateException("standard SHA-256 unavailable");}
+ }
+ private static ReceiverException rejected(String code){return new ReceiverException(code);}
+ @Override public String toString(){return "SqliteCommittedReceiver[authority=]";}
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedSnapshot.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedSnapshot.java
new file mode 100644
index 00000000..8647210d
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedSnapshot.java
@@ -0,0 +1,227 @@
+package io.teaql.core.sqlite;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.teaql.core.UserContext;
+import io.teaql.core.meta.EntityDescriptor;
+import io.teaql.core.meta.EntityMetaFactory;
+import io.teaql.core.meta.PropertyDescriptor;
+import io.teaql.core.meta.Relation;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
+import static io.teaql.core.sqlite.SqliteModelCheckedCommittedTarget.*;
+
+/** Trusted startup infrastructure. No ordinary business save or remote request routing. */
+public final class SqliteCommittedSnapshot {
+ private static final Set SNAPSHOT_FIELDS = Set.of("snapshot_version", "snapshot_id",
+ "source_runtime_id", "source_sequence", "source_model_hash", "audit", "entity_types", "rows");
+ private static final Set AUDIT_FIELDS = Set.of("actor_ref", "comment", "purpose", "trace_id");
+ private static final ObjectMapper JSON = new ObjectMapper()
+ .enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION)
+ .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
+ public record Limits(int maxRows, int maxPayloadBytes) {
+ public Limits { if (maxRows < 1 || maxRows == Integer.MAX_VALUE || maxPayloadBytes < 1)
+ throw new IllegalArgumentException("positive bounded snapshot limits are required"); }
+ public static Limits defaults() { return new Limits(100_000, 16 * 1024 * 1024); }
+ }
+ public enum ImportMode { REQUIRE_EMPTY, REPLACE_UNINITIALIZED }
+ public record SnapshotRecord(String sourceRuntimeId, long sourceSequence, String snapshotId,
+ String modelHash, String payloadSha256, String payloadJson) {
+ @Override public String toString() { return "SnapshotRecord[sourceSequence=" + sourceSequence + ", payload=]"; }
+ }
+ private record Table(EntityDescriptor entity, List fields,
+ String table, String id, String version) { }
+ private final SqlExecutionAdapter sql;
+ private final String source;
+ private final String model;
+
+ public SqliteCommittedSnapshot(SqlExecutionAdapter sql, String source, String model) {
+ this.sql = Objects.requireNonNull(sql, "sql");
+ if (blank(source) || model == null || !model.matches("sha256:[0-9a-f]{64}")) throw invalid("source/model is invalid");
+ this.source = source; this.model = model;
+ }
+
+ /** N and physical rows share one read transaction. Never export a caller's uncommitted writes. */
+ public SnapshotRecord exportSnapshot(UserContext context, String snapshotId,
+ String comment, String purpose, Limits limits) {
+ Objects.requireNonNull(context, "context"); Objects.requireNonNull(limits, "limits");
+ if (blank(snapshotId) || blank(comment) || blank(purpose)) throw invalid("snapshot identity and intent are required");
+ standalone();
+ EntityMetaFactory metadata = EntityMetaFactory.requireFrom(context);
+ AtomicReference result = new AtomicReference<>();
+ sql.executeInTransaction(() -> {
+ requireTransaction();
+ // The first SELECT establishes SQLite's read snapshot, including when N=0.
+ long n = position();
+ List tables = tables(metadata);
+ ObjectNode body = JSON.createObjectNode();
+ body.put("snapshot_version", 1); body.put("snapshot_id", snapshotId);
+ body.put("source_runtime_id", source); body.put("source_sequence", Long.toString(n));
+ body.put("source_model_hash", model);
+ body.putObject("audit").put("comment", comment).put("purpose", purpose);
+ var catalog = body.putArray("entity_types"); var rows = body.putArray("rows");
+ long usedBytes = 0;
+ for (Table table : tables) {
+ catalog.add(table.entity().getType());
+ List columns = new ArrayList<>(List.of(quoted(table.id()) + " AS snapshot_id", quoted(table.version()) + " AS snapshot_version"));
+ for (int i = 0; i < table.fields().size(); i++) columns.add(quoted(mapping(table.fields().get(i)).column()) + " AS snapshot_field_" + i);
+ int remaining = limits.maxRows() - rows.size();
+ for (Map values : sql.queryForList("SELECT " + String.join(",", columns)
+ + " FROM " + quoted(table.table()) + " ORDER BY " + quoted(table.id()) + " LIMIT ?",
+ new Object[] {remaining + 1})) {
+ if (rows.size() == limits.maxRows()) throw invalid("snapshot row budget exceeded");
+ ObjectNode row = JSON.createObjectNode().put("entity_type", table.entity().getType())
+ .put("entity_id", integer(values.get("snapshot_id")).toString())
+ .put("version", integer(values.get("snapshot_version")).toString());
+ ObjectNode fields = row.putObject("fields");
+ for (int i = 0; i < table.fields().size(); i++) fields.set(table.fields().get(i).getName(), encode(table.fields().get(i), values.get("snapshot_field_" + i)));
+ // Same structural decoder/representability checks as the target, not Checker/Fix.
+ decodeSnapshotRow(metadata, row);
+ usedBytes += bytes(row.toString()); if (usedBytes > limits.maxPayloadBytes()) throw invalid("snapshot byte budget exceeded");
+ rows.add(row);
+ }
+ }
+ String text = body.toString(); bounded(text, limits);
+ result.set(new SnapshotRecord(source, n, snapshotId, model, digest(text), text));
+ });
+ return result.get();
+ }
+
+ /** Explicit destructive replacement is allowed only before this target has any stream state. */
+ public SqliteCommittedEventStore.InboxOutcome initializeSnapshot(UserContext context, SnapshotRecord record, Limits limits) {
+ return initializeSnapshot(context, record, ImportMode.REQUIRE_EMPTY, limits);
+ }
+
+ /** Explicit destructive replacement is allowed only before this target has any stream state. */
+ public SqliteCommittedEventStore.InboxOutcome initializeSnapshot(UserContext context,
+ SnapshotRecord record, ImportMode mode, Limits limits) {
+ Objects.requireNonNull(context, "context"); Objects.requireNonNull(mode, "mode"); Objects.requireNonNull(limits, "limits");
+ sql.restrictOrdinaryMutations();
+ standalone(); JsonNode body = validate(record, limits);
+ EntityMetaFactory metadata = EntityMetaFactory.requireFrom(context);
+ AtomicReference result = new AtomicReference<>();
+ sql.executeInTransaction(() -> {
+ requireTransaction();
+ List> prior = sql.queryForList("SELECT snapshot_id,source_sequence,model_hash,payload_sha256,payload_json "
+ + "FROM teaql_event_snapshot WHERE source_runtime_id=?", new Object[] {source});
+ if (!prior.isEmpty()) {
+ Map old = prior.get(0);
+ if (!record.snapshotId().equals(old.get("snapshot_id")) || record.sourceSequence() != integer(old.get("source_sequence"))
+ || !model.equals(old.get("model_hash")) || !record.payloadSha256().equals(old.get("payload_sha256"))
+ || !record.payloadSha256().equals(digest((String) old.get("payload_json")))) throw conflict("snapshot identity/content conflicts");
+ var checkpoint = sql.queryForList("SELECT source_sequence,model_hash FROM teaql_event_checkpoint WHERE source_runtime_id=?", new Object[] {source});
+ if (checkpoint.size() != 1 || integer(checkpoint.get(0).get("source_sequence")) < record.sourceSequence()
+ || !model.equals(checkpoint.get(0).get("model_hash"))) throw invalid("snapshot checkpoint is inconsistent");
+ result.set(SqliteCommittedEventStore.InboxOutcome.duplicate(record.sourceSequence())); return;
+ }
+ for (String table : List.of("teaql_event_snapshot", "teaql_event_checkpoint", "teaql_event_inbox", "teaql_event_applied_payload", "teaql_event_outbox", "teaql_event_sequence", "teaql_event_snapshot_publication", "teaql_event_delivery_journal"))
+ if (count(table) != 0) throw conflict("target already contains stream state");
+ List tables = tables(metadata);
+ Set catalog = new TreeSet<>();
+ if (!body.path("entity_types").isArray()) throw invalid("snapshot catalog is missing");
+ for (JsonNode type : body.path("entity_types")) if (!type.isTextual() || !catalog.add(type.textValue())) throw invalid("snapshot catalog is invalid");
+ if (!catalog.equals(new TreeSet<>(tables.stream().map(t -> t.entity().getType()).toList()))) throw invalid("snapshot catalog differs from model");
+ List mutations = new ArrayList<>(); Set identities = new HashSet<>();
+ for (JsonNode row : body.path("rows")) {
+ Mutation mutation = decodeSnapshotRow(metadata, row);
+ if (!identities.add(mutation.table() + ":" + mutation.id())) throw invalid("duplicate snapshot row");
+ mutations.add(mutation);
+ }
+ sql.execute("PRAGMA defer_foreign_keys=ON");
+ for (Table table : tables) {
+ if (mode == ImportMode.REQUIRE_EMPTY && count(table.table()) != 0) throw conflict("target domain is not empty; explicit replacement required");
+ if (mode == ImportMode.REPLACE_UNINITIALIZED) sql.execute("DELETE FROM " + quoted(table.table()));
+ }
+ for (Mutation mutation : mutations) applyOne(sql, mutation);
+ // Later triggers/rows must not rewrite earlier authoritative facts.
+ for (Mutation mutation : mutations) verifyReadback(sql, mutation);
+ for (Table table : tables) {
+ long expected = mutations.stream().filter(m -> m.table().equals(table.table())).count();
+ if (count(table.table()) != expected) throw invalid("target snapshot row count differs");
+ if (!sql.queryForList("PRAGMA foreign_key_check(" + quoted(table.table()) + ")", new Object[0]).isEmpty()) throw invalid("snapshot has a dangling reference");
+ }
+ sql.update("INSERT INTO teaql_event_snapshot(source_runtime_id,snapshot_id,source_sequence,model_hash,payload_sha256,payload_json) VALUES (?,?,?,?,?,?)",
+ new Object[] {source, record.snapshotId(), record.sourceSequence(), model, record.payloadSha256(), record.payloadJson()});
+ sql.update("INSERT INTO teaql_event_checkpoint(source_runtime_id,source_sequence,model_hash) VALUES (?,?,?)", new Object[] {source, record.sourceSequence(), model});
+ result.set(SqliteCommittedEventStore.InboxOutcome.applied(record.sourceSequence()));
+ });
+ return result.get();
+ }
+
+ JsonNode validate(SnapshotRecord record, Limits limits) {
+ if (record == null || !source.equals(record.sourceRuntimeId())) throw invalid("snapshot source is not authorized");
+ if (!model.equals(record.modelHash())) throw invalid("snapshot model mismatch");
+ if (blank(record.snapshotId()) || record.sourceSequence() < 0) throw invalid("snapshot identity/position is invalid");
+ bounded(record.payloadJson(), limits);
+ if (!digest(record.payloadJson()).equals(record.payloadSha256())) throw invalid("snapshot digest mismatch");
+ JsonNode body;
+ try { body = JSON.readTree(record.payloadJson()); } catch (Exception failure) { throw invalid("snapshot JSON is invalid"); }
+ if (!body.isObject() || body.size() != 8 || !body.path("snapshot_version").isInt() || body.path("snapshot_version").intValue() != 1
+ || !record.snapshotId().equals(body.path("snapshot_id").textValue()) || !source.equals(body.path("source_runtime_id").textValue())
+ || !Long.toString(record.sourceSequence()).equals(body.path("source_sequence").textValue()) || !model.equals(body.path("source_model_hash").textValue())
+ || !body.path("rows").isArray() || body.path("rows").size() > limits.maxRows()
+ || blank(body.path("audit").path("comment").asText()) || blank(body.path("audit").path("purpose").asText())) throw invalid("snapshot header or rows are invalid");
+ SqliteCommittedEventStore.rejectUnknownControls(body, SNAPSHOT_FIELDS);
+ SqliteCommittedEventStore.rejectUnknownControls(body.path("audit"), AUDIT_FIELDS);
+ return body;
+ }
+ private long position() {
+ var positions = sql.queryForList("SELECT current_sequence FROM teaql_event_sequence WHERE source_runtime_id=?", new Object[] {source});
+ long n = positions.isEmpty() ? 0 : integer(positions.get(0).get("current_sequence"));
+ var latest = sql.queryForList("SELECT source_sequence,model_hash FROM teaql_event_outbox WHERE source_runtime_id=? ORDER BY source_sequence DESC LIMIT 1", new Object[] {source});
+ if (n < 0 || n == 0 && !latest.isEmpty() || n > 0 && (latest.size() != 1 || integer(latest.get(0).get("source_sequence")) != n || !model.equals(latest.get(0).get("model_hash")))) throw invalid("source sequence/model is inconsistent");
+ return n;
+ }
+ private static List tables(EntityMetaFactory metadata) {
+ List result = new ArrayList<>(); Set names = new HashSet<>(), physical = new HashSet<>();
+ for (EntityDescriptor entity : metadata.allEntityDescriptors()) {
+ if (entity.getParent() != null || entity.hasChildren() || entity.findIdProperty() == null || entity.findVersionProperty() == null) throw invalid("unsupported snapshot entity mapping");
+ Mapping id = mapping(entity.findIdProperty()), version = mapping(entity.findVersionProperty());
+ String physicalName = id.table().toLowerCase(Locale.ROOT);
+ if (!names.add(entity.getType()) || !physical.add(physicalName) || physicalName.startsWith("teaql_event_") || !id.table().equals(version.table())) throw invalid("ambiguous/reserved snapshot table");
+ var fields = entity.getOwnProperties().stream().filter(p -> !p.isId() && !p.isVersion()).toList();
+ Set columns = new HashSet<>(List.of(id.column(), version.column()));
+ for (PropertyDescriptor field : fields) if (!id.table().equals(mapping(field).table()) || !columns.add(mapping(field).column())) throw invalid("unsupported snapshot field mapping");
+ result.add(new Table(entity, fields, id.table(), id.column(), version.column()));
+ }
+ if (result.isEmpty()) throw invalid("snapshot requires an installed model");
+ result.sort(Comparator.comparing(t -> t.entity().getType())); return result;
+ }
+ private static JsonNode encode(PropertyDescriptor property, Object value) {
+ JsonNode wire;
+ Class type = property.getType() == null ? null : property.getType().javaType();
+ if (value == null) wire = JSON.nullNode();
+ else if (property instanceof Relation relation) wire = JSON.createObjectNode()
+ .put("entity_type", relation.getReverseProperty().getOwner().getType()).put("entity_id", integer(value).toString());
+ else if (type == Boolean.class || type == boolean.class) {
+ long number = integer(value); if (number != 0 && number != 1) throw invalid("source Boolean storage is invalid"); wire = JSON.valueToTree(number == 1);
+ } else if (type == BigDecimal.class || type == java.math.BigInteger.class) {
+ if (value instanceof Double || value instanceof Float) throw invalid("source Decimal is approximate REAL"); wire = JSON.valueToTree(value.toString());
+ } else if (type == Float.class || type == float.class || type == Double.class || type == double.class) {
+ if (!(value instanceof Number number) || !Double.isFinite(number.doubleValue())) throw invalid("source floating storage is invalid"); wire = JSON.valueToTree(number.doubleValue());
+ } else if (type == byte[].class && value instanceof byte[] bytes) wire = JSON.createObjectNode().put("$binary_base64", Base64.getEncoder().encodeToString(bytes));
+ else if (value instanceof Byte || value instanceof Short || value instanceof Integer || value instanceof Long) wire = JSON.valueToTree(integer(value).toString());
+ else if (value instanceof String text) wire = JSON.valueToTree(text);
+ else throw invalid("source storage representation is unsupported");
+ Object decoded = decodeValue(property, wire);
+ if (!sameValue(new Field(mapping(property).column(), decoded, type == BigDecimal.class || type == java.math.BigInteger.class), value)) throw invalid("source field cannot be represented losslessly");
+ return wire;
+ }
+ private long count(String table) { return integer(sql.queryForList("SELECT COUNT(*) AS n FROM " + quoted(table), new Object[0]).get(0).get("n")); }
+ private void standalone() { if (sql.hasActiveTransaction()) throw invalid("snapshot requires a standalone transaction"); }
+ private void requireTransaction() { if (!sql.hasActiveTransaction()) throw invalid("snapshot adapter must provide an active SQL transaction"); }
+ private static Long integer(Object value) { if (!(value instanceof Long || value instanceof Integer || value instanceof Short || value instanceof Byte)) throw invalid("source identity/integer storage is invalid"); return ((Number) value).longValue(); }
+ private static int bytes(String text) { return text.getBytes(StandardCharsets.UTF_8).length; }
+ private static void bounded(String text, Limits limits) { if (text == null || text.length() > limits.maxPayloadBytes() || bytes(text) > limits.maxPayloadBytes()) throw invalid("snapshot byte budget exceeded"); }
+ private static String digest(String text) { try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(text.getBytes(StandardCharsets.UTF_8))); } catch (java.security.NoSuchAlgorithmException impossible) { throw new AssertionError(impossible); } }
+ private static boolean blank(String text) { return text == null || text.isBlank(); }
+ private static SqliteCommittedEventStore.CommittedEventException invalid(String message) { return new SqliteCommittedEventStore.CommittedEventException("INVALID_SNAPSHOT", message); }
+ private static SqliteCommittedEventStore.CommittedEventException conflict(String message) { return new SqliteCommittedEventStore.CommittedEventException("SNAPSHOT_CONFLICT", message); }
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedSnapshotPublisher.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedSnapshotPublisher.java
new file mode 100644
index 00000000..040d9e23
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedSnapshotPublisher.java
@@ -0,0 +1,125 @@
+package io.teaql.core.sqlite;
+
+import io.teaql.core.UserContext;
+import io.teaql.core.meta.EntityMetaFactory;
+import io.teaql.core.sqlite.SqliteCommittedSnapshot.Limits;
+import io.teaql.core.sqlite.SqliteCommittedSnapshot.SnapshotRecord;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Supplier;
+
+/** Durable startup bootstrap for one trusted source/target slot, not a remote API. */
+public final class SqliteCommittedSnapshotPublisher {
+ public record PreparedSnapshot(SnapshotRecord snapshot, boolean acknowledged) { }
+ /** Correlation, not authentication: the transport must authenticate its peer. */
+ public record Receipt(String sourceRuntimeId, String targetId, String snapshotId,
+ long sourceSequence, String modelHash, String payloadSha256, boolean duplicate) { }
+ @FunctionalInterface public interface SnapshotSink { Receipt deliver(SnapshotRecord record); }
+
+ private final SqlExecutionAdapter sql;
+ private final String source, target, model;
+ private final SqliteCommittedSnapshot snapshots;
+ public SqliteCommittedSnapshotPublisher(SqlExecutionAdapter sql, String source, String target, String model) {
+ this.sql = Objects.requireNonNull(sql, "sql");
+ if (target == null || target.isBlank()) throw failure("INVALID_SNAPSHOT", "target identity is required");
+ this.source = source; this.target = target; this.model = model;
+ this.snapshots = new SqliteCommittedSnapshot(sql, source, model);
+ }
+
+ /** Retain original bytes before delivery; retries never re-export a published identity. */
+ public PreparedSnapshot prepare(UserContext context, String snapshotId, String comment, String purpose, Limits limits) {
+ require(context, snapshotId, limits);
+ if (comment == null || comment.isBlank() || purpose == null || purpose.isBlank())
+ throw failure("INVALID_SNAPSHOT", "snapshot intent is required");
+ PreparedSnapshot prior = transaction(() -> existing(snapshotId, limits));
+ if (prior != null) return prior;
+ SnapshotRecord captured = snapshots.exportSnapshot(context, snapshotId, comment, purpose, limits);
+ return transaction(() -> {
+ PreparedSnapshot raced = existing(snapshotId, limits);
+ if (raced != null) return raced; // First durable record wins, never replace with a newer view.
+ if (!cursor().isEmpty()) throw failure("CURSOR_CONFLICT", "target stream has already started");
+ sql.update("INSERT INTO teaql_event_snapshot_publication "
+ + "(source_runtime_id,target_id,snapshot_id,source_sequence,model_hash,payload_sha256,payload_json,acknowledged) "
+ + "VALUES (?,?,?,?,?,?,?,0)", new Object[] {source,target,captured.snapshotId(),captured.sourceSequence(),model,captured.payloadSha256(),captured.payloadJson()});
+ return new PreparedSnapshot(captured, false);
+ });
+ }
+
+ /** One attempt. Lost or mismatching ACK leaves the retained job/cursor untouched. */
+ public PreparedSnapshot publish(UserContext context, String snapshotId, Limits limits, SnapshotSink sink) {
+ require(context, snapshotId, limits); Objects.requireNonNull(sink, "sink");
+ PreparedSnapshot prepared = transaction(() -> existing(snapshotId, limits));
+ if (prepared == null) throw failure("INVALID_SNAPSHOT", "snapshot has not been durably prepared");
+ if (prepared.acknowledged()) return prepared;
+ SnapshotRecord record = prepared.snapshot();
+ var journal = new SqliteCommittedObservation(sql,source,target,model);
+ var attempt = journal.begin("snapshot",record.sourceSequence(),record.snapshotId(),record.payloadSha256());
+ Receipt receipt;
+ try {
+ receipt = sink.deliver(record); // No source SQL transaction/connection held across transport.
+ if (receipt == null || !source.equals(receipt.sourceRuntimeId()) || !target.equals(receipt.targetId())
+ || !record.snapshotId().equals(receipt.snapshotId()) || record.sourceSequence() != receipt.sourceSequence()
+ || !model.equals(receipt.modelHash()) || !record.payloadSha256().equals(receipt.payloadSha256()))
+ throw failure("ACK_MISMATCH", "snapshot receipt does not match this source/target record");
+ } catch (RuntimeException error) { journal.failed(attempt,error); throw error; }
+ return journal.acknowledged(attempt, () -> {
+ PreparedSnapshot current = existing(snapshotId, limits);
+ if (current == null || !current.snapshot().equals(record)) throw failure("SNAPSHOT_CONFLICT", "prepared snapshot changed");
+ if (current.acknowledged()) return current; // Tail may already have advanced; never rewind.
+ if (!cursor().isEmpty()) throw failure("CURSOR_CONFLICT", "bootstrap cursor changed concurrently");
+ var status = new SqliteCommittedOutboxPublisher(sql,source,target,model).status();
+ if (status.sourceHighWatermark() < record.sourceSequence()) throw failure("INVALID_SNAPSHOT", "snapshot is ahead of source");
+ sql.update("INSERT INTO teaql_event_delivery_cursor(source_runtime_id,target_id,acked_sequence,model_hash) VALUES (?,?,?,?)",
+ new Object[] {source,target,record.sourceSequence(),model});
+ if (sql.update("UPDATE teaql_event_snapshot_publication SET acknowledged=1 WHERE source_runtime_id=? AND target_id=? "
+ + "AND snapshot_id=? AND payload_sha256=? AND acknowledged=0", new Object[] {source,target,snapshotId,record.payloadSha256()}) != 1)
+ throw failure("CURSOR_CONFLICT", "bootstrap acknowledgement changed concurrently");
+ return new PreparedSnapshot(record, true);
+ });
+ }
+
+ /** Target helper: construct a receipt only after the authoritative target transaction commits. */
+ public static Receipt initializeAndAcknowledge(UserContext context, SqliteCommittedSnapshot target,
+ SnapshotRecord record, String targetId, Limits limits) {
+ return initializeAndAcknowledge(context,target,record,targetId,SqliteCommittedSnapshot.ImportMode.REQUIRE_EMPTY,limits);
+ }
+ public static Receipt initializeAndAcknowledge(UserContext context, SqliteCommittedSnapshot target,
+ SnapshotRecord record, String targetId, SqliteCommittedSnapshot.ImportMode mode, Limits limits) {
+ if (targetId == null || targetId.isBlank()) throw failure("INVALID_SNAPSHOT", "target identity is required");
+ var outcome = target.initializeSnapshot(context,record,mode,limits);
+ if (outcome.checkpoint() != record.sourceSequence()) throw failure("ACK_MISMATCH", "target snapshot position differs");
+ return new Receipt(record.sourceRuntimeId(),targetId,record.snapshotId(),record.sourceSequence(),record.modelHash(),record.payloadSha256(),outcome.duplicate());
+ }
+
+ private PreparedSnapshot existing(String snapshotId, Limits limits) {
+ var rows = sql.queryForList("SELECT snapshot_id,source_sequence,model_hash,payload_sha256,payload_json,acknowledged "
+ + "FROM teaql_event_snapshot_publication WHERE source_runtime_id=? AND target_id=?", new Object[] {source,target});
+ if (rows.isEmpty()) return null;
+ var row = rows.get(0);
+ if (!snapshotId.equals(row.get("snapshot_id"))) throw failure("SNAPSHOT_CONFLICT", "target slot already has a different snapshot");
+ var record = new SnapshotRecord(source,((Number)row.get("source_sequence")).longValue(),snapshotId,
+ (String)row.get("model_hash"),(String)row.get("payload_sha256"),(String)row.get("payload_json"));
+ snapshots.validate(record,limits);
+ int flag = ((Number)row.get("acknowledged")).intValue();
+ if (flag != 0 && flag != 1) throw failure("INVALID_SNAPSHOT", "invalid acknowledgement state");
+ boolean acked = flag == 1;
+ var cursors = cursor();
+ if (!acked && !cursors.isEmpty() || acked && (cursors.size()!=1 || !model.equals(cursors.get(0).get("model_hash"))
+ || ((Number)cursors.get(0).get("acked_sequence")).longValue()> cursor() { return sql.queryForList(
+ "SELECT acked_sequence,model_hash FROM teaql_event_delivery_cursor WHERE source_runtime_id=? AND target_id=?",new Object[]{source,target}); }
+ private void require(UserContext context, String snapshotId, Limits limits) {
+ Objects.requireNonNull(context,"context"); Objects.requireNonNull(limits,"limits");
+ EntityMetaFactory.requireFrom(context);
+ if (snapshotId == null || snapshotId.isBlank()) throw failure("INVALID_SNAPSHOT","snapshot identity is required");
+ if (sql.hasActiveTransaction()) throw failure("INVALID_SNAPSHOT","bootstrap requires a standalone transaction");
+ }
+ private T transaction(Supplier action) { AtomicReference value=new AtomicReference<>();sql.executeInTransaction(()->{
+ if (!sql.hasActiveTransaction()) throw failure("INVALID_SNAPSHOT","bootstrap adapter must provide an active SQL transaction");
+ value.set(action.get());});return value.get(); }
+ private static SqliteCommittedEventStore.CommittedEventException failure(String code,String message) { return new SqliteCommittedEventStore.CommittedEventException(code,message); }
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteModelCheckedCommittedTarget.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteModelCheckedCommittedTarget.java
new file mode 100644
index 00000000..7abb3316
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteModelCheckedCommittedTarget.java
@@ -0,0 +1,445 @@
+package io.teaql.core.sqlite;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.teaql.core.UserContext;
+import io.teaql.core.meta.EntityDescriptor;
+import io.teaql.core.meta.EntityMetaFactory;
+import io.teaql.core.meta.PropertyDescriptor;
+import io.teaql.core.meta.Relation;
+import io.teaql.core.sql.SQLColumn;
+import io.teaql.core.sql.portable.SQLPropertyUtil;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+
+import java.math.BigDecimal;
+import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Startup-owned, one-source SQLite target for committed model facts. This is
+ * infrastructure, not a business-facing Mutation or remote query endpoint.
+ */
+public final class SqliteModelCheckedCommittedTarget {
+ private static final Set AUDIT_FIELDS = Set.of("actor_ref", "comment", "purpose", "trace_id");
+ private static final Set MUTATION_FIELDS = Set.of("entity_type", "entity_id", "operation",
+ "base_version", "committed_version", "fields", "trace_chain");
+ private static final Set TRACE_KINDS = Set.of(
+ "OPERATION", "REQUEST", "RELATION", "ENTITY", "PROVIDER", "SQL",
+ "COMMENT", "PURPOSE", "AUDIT_REASON");
+ private static final Pattern DECLARED_DECIMAL = Pattern.compile(
+ "(?i)\\s*(?:NUMERIC|DECIMAL)\\s*\\(\\s*([0-9]+)\\s*,\\s*([0-9]+)\\s*\\)\\s*");
+ private final SqlExecutionAdapter sql;
+ private final SqliteCommittedEventStore events;
+ private final String sourceRuntimeId;
+ private final String modelHash;
+
+ public SqliteModelCheckedCommittedTarget(SqlExecutionAdapter sql,
+ String sourceRuntimeId, String modelHash) {
+ this.sql = Objects.requireNonNull(sql, "sql");
+ if (sourceRuntimeId == null || sourceRuntimeId.isBlank()
+ || modelHash == null || !modelHash.matches("sha256:[0-9a-f]{64}")) {
+ throw invalid("target source identity or SHA-256 model hash is invalid");
+ }
+ this.sourceRuntimeId = sourceRuntimeId;
+ this.modelHash = modelHash;
+ sql.restrictOrdinaryMutations();
+ this.events = new SqliteCommittedEventStore(sql);
+ }
+
+ public void ensureSchema() { events.ensureSchema(); }
+
+ /** One graph, its Inbox identity, original audit payload and checkpoint commit together. */
+ public SqliteCommittedEventStore.InboxOutcome apply(
+ UserContext context, SqliteCommittedEventStore.OutboxRecord record) {
+ Objects.requireNonNull(context, "context");
+ if (record == null) throw invalid("committed event is missing");
+ if (!sourceRuntimeId.equals(record.sourceRuntimeId())) {
+ throw new SqliteCommittedEventStore.CommittedEventException(
+ "UNAUTHORIZED_SOURCE", "committed-event source is not authorized");
+ }
+ // Resolve from this Context, never from the legacy process-global fallback.
+ EntityMetaFactory metadata = EntityMetaFactory.requireFrom(context);
+ AtomicReference result = new AtomicReference<>();
+ sql.executeInTransaction(() -> result.set(events.apply(record, modelHash,
+ (transaction, payload) -> applyGraph(transaction, metadata, payload))));
+ return result.get();
+ }
+
+ private static void applyGraph(SqlExecutionAdapter sql, EntityMetaFactory metadata, JsonNode payload) {
+ JsonNode audit = payload.path("audit");
+ SqliteCommittedEventStore.rejectUnknownControls(audit, AUDIT_FIELDS);
+ if (!audit.isObject() || !nonblank(audit.path("comment"))
+ || !nonblank(audit.path("purpose"))
+ || !nonblank(payload.path("source_transaction_id"))) {
+ throw invalid("source transaction or audit intent is missing");
+ }
+ try { OffsetDateTime.parse(payload.path("committed_at").asText()); }
+ catch (RuntimeException failure) { throw invalid("commit time is invalid"); }
+ List mutations = new ArrayList<>();
+ Set seen = new HashSet<>();
+ for (JsonNode fact : payload.path("mutations")) {
+ Mutation mutation = decode(metadata, fact);
+ if (!seen.add(mutation.entityType() + ":" + mutation.id())) {
+ throw invalid("one set mutates the same entity twice");
+ }
+ mutations.add(mutation);
+ }
+ // Decode every fact before touching domain rows. The surrounding
+ // savepoint also rolls back a later FK or readback failure.
+ for (Mutation mutation : mutations) applyOne(sql, mutation);
+ }
+
+ private static Mutation decode(EntityMetaFactory metadata, JsonNode fact) {
+ return decode(metadata, fact, false);
+ }
+
+ /** Package-private physical restore mode; never accepted by ordinary event apply. */
+ static Mutation decodeSnapshotRow(EntityMetaFactory metadata, JsonNode row) {
+ if (!row.isObject() || row.size() != 4 || !row.has("entity_type")
+ || !row.has("entity_id") || !row.has("version") || !row.has("fields")) {
+ throw invalid("snapshot row has wrong shape");
+ }
+ long id = positiveId(row.path("entity_id"));
+ long version = signedVersion(row.path("version"));
+ if (!Long.toString(id).equals(row.path("entity_id").asText())
+ || !Long.toString(version).equals(row.path("version").asText())
+ || version == 0 || version == Long.MIN_VALUE) throw invalid("snapshot identity/version is invalid");
+ var fact = row.deepCopy();
+ var object = (com.fasterxml.jackson.databind.node.ObjectNode) fact;
+ object.remove("version"); object.put("committed_version", Long.toString(version));
+ object.put("operation", "create"); object.putNull("base_version"); object.putArray("trace_chain");
+ return decode(metadata, object, true);
+ }
+
+ private static Mutation decode(EntityMetaFactory metadata, JsonNode fact, boolean snapshot) {
+ if (!fact.isObject() || !nonblank(fact.path("entity_type"))) {
+ throw invalid("mutation entity type is missing");
+ }
+ SqliteCommittedEventStore.rejectUnknownControls(fact, MUTATION_FIELDS);
+ validateTraceChain(fact.path("trace_chain"));
+ String entityType = fact.path("entity_type").textValue();
+ EntityDescriptor entity = metadata.allEntityDescriptors().stream()
+ .filter(candidate -> entityType.equals(candidate.getType())).findFirst()
+ .orElseThrow(() -> invalid("unknown entity type"));
+ // Multi-table inheritance needs an explicit target mapping. Reject it
+ // instead of silently writing only the leaf table.
+ if (entity.getParent() != null || entity.hasChildren()) {
+ throw invalid("multi-table entity target mapping is not supported");
+ }
+ PropertyDescriptor idProperty = entity.findIdProperty();
+ PropertyDescriptor versionProperty = entity.findVersionProperty();
+ if (idProperty == null || versionProperty == null) {
+ throw invalid("entity has no ID or version property");
+ }
+ Mapping idMapping = mapping(idProperty);
+ Mapping versionMapping = mapping(versionProperty);
+ if (!idMapping.table().equals(versionMapping.table())) {
+ throw invalid("identity and version span multiple tables");
+ }
+ long id = positiveId(fact.path("entity_id"));
+ String operation = fact.path("operation").asText("");
+ if (!List.of("create", "update", "delete", "recover").contains(operation)) {
+ throw invalid("unknown mutation operation");
+ }
+ JsonNode baseNode = fact.path("base_version");
+ Long base = "create".equals(operation) && baseNode.isNull()
+ ? null : signedVersion(baseNode);
+ long committed = signedVersion(fact.path("committed_version"));
+ boolean valid = switch (operation) {
+ case "create" -> base == null && (snapshot ? committed != 0 && committed != Long.MIN_VALUE : committed == 1);
+ case "update" -> base != null && base > 0 && base < Long.MAX_VALUE
+ && committed == base + 1;
+ case "delete" -> base != null && base > 0 && base < Long.MAX_VALUE
+ && committed == -(base + 1);
+ case "recover" -> base != null && base < 0 && base != Long.MIN_VALUE
+ && -base < Long.MAX_VALUE && committed == -base + 1;
+ default -> false;
+ };
+ if (!valid) throw invalid("invalid version transition");
+ JsonNode supplied = fact.path("fields");
+ if (!supplied.isObject()) throw invalid("mutation fields are missing");
+ if ("delete".equals(operation) && !supplied.isEmpty()) {
+ throw invalid("delete must be a tombstone");
+ }
+ List ownProperties = entity.getOwnProperties();
+ supplied.fieldNames().forEachRemaining(name -> {
+ PropertyDescriptor property = entity.findProperty(name);
+ if (property == null || !ownProperties.contains(property)
+ || property.isId() || property.isVersion()) {
+ throw invalid("unknown or reserved model field");
+ }
+ });
+ List fields = new ArrayList<>();
+ for (PropertyDescriptor property : ownProperties) {
+ if (property.isId() || property.isVersion()) continue;
+ JsonNode value = supplied.get(property.getName());
+ if (value == null) {
+ if ("create".equals(operation)) throw invalid("create omitted a modeled field");
+ continue;
+ }
+ Mapping column = mapping(property);
+ if (!idMapping.table().equals(column.table())) {
+ throw invalid("mutation field spans multiple tables");
+ }
+ Class type = property.getType() == null ? null : property.getType().javaType();
+ fields.add(new Field(column.column(), decodeValue(property, value),
+ type == BigDecimal.class || type == java.math.BigInteger.class));
+ }
+ return new Mutation(entityType, idMapping.table(), idMapping.column(),
+ versionMapping.column(), id, base, committed, operation, fields);
+ }
+
+ private static void validateTraceChain(JsonNode chain) {
+ if (!chain.isArray()) throw invalid("mutation trace chain is missing");
+ for (JsonNode node : chain) {
+ JsonNode id = node.path("entity_id");
+ if (!node.isObject() || node.size() != 4
+ || !TRACE_KINDS.contains(node.path("kind").asText(""))
+ || !node.path("name").isTextual()
+ || !node.path("comment").isTextual()
+ || !(id.isNull() || id.isTextual() && id.textValue().matches("[1-9][0-9]*"))) {
+ throw invalid("trace node has wrong shape");
+ }
+ }
+ }
+
+ static Object decodeValue(PropertyDescriptor property, JsonNode value) {
+ if (value.isNull()) {
+ if (property.getBoolean("required", false)) throw invalid("non-null field is null");
+ return null;
+ }
+ if (property instanceof Relation relation) {
+ if (!value.isObject() || value.size() != 2 || relation.getReverseProperty() == null
+ || relation.getReverseProperty().getOwner() == null
+ || !relation.getReverseProperty().getOwner().getType()
+ .equals(value.path("entity_type").asText())) {
+ throw invalid("reference target does not match model");
+ }
+ return positiveId(value.path("entity_id"));
+ }
+ Class type = property.getType() == null ? null : property.getType().javaType();
+ if (type == java.time.LocalDate.class || type == java.sql.Date.class) {
+ if (!value.isTextual()) throw invalid("Date field has wrong type");
+ String text = value.textValue();
+ try {
+ if (!java.time.LocalDate.parse(text).toString().equals(text)) {
+ throw invalid("Date field has wrong type");
+ }
+ } catch (java.time.format.DateTimeParseException failure) {
+ throw invalid("Date field has wrong type");
+ }
+ return text;
+ }
+ if (type == java.time.Instant.class || type == java.util.Date.class
+ || type == java.sql.Timestamp.class) {
+ if (!value.isTextual() || !value.textValue().matches("-?(0|[1-9][0-9]*)")
+ || value.textValue().equals("-0")) {
+ throw invalid("Timestamp field has wrong type");
+ }
+ try { return Long.parseLong(value.textValue()); }
+ catch (NumberFormatException failure) { throw invalid("Timestamp field has wrong type"); }
+ }
+ if (type == java.time.LocalDateTime.class) {
+ throw invalid("zone-free LocalDateTime has no committed instant mapping");
+ }
+ if (type == String.class || type == java.util.UUID.class || type != null && type.isEnum()) {
+ if (!value.isTextual()) throw invalid("text field has wrong type");
+ return value.textValue();
+ }
+ if (type == Boolean.class || type == boolean.class) {
+ if (!value.isBoolean()) throw invalid("Boolean field has wrong type");
+ return value.booleanValue() ? 1 : 0;
+ }
+ if (type == Long.class || type == long.class || type == Integer.class || type == int.class
+ || type == Short.class || type == short.class || type == Byte.class || type == byte.class) {
+ if (!value.isTextual() || !value.textValue().matches("-?(0|[1-9][0-9]*)")
+ || value.textValue().equals("-0")) {
+ throw invalid("integer field has wrong type");
+ }
+ long number;
+ try { number = Long.parseLong(value.textValue()); }
+ catch (NumberFormatException failure) { throw invalid("integer field overflows"); }
+ if (type == Integer.class || type == int.class) {
+ if (number < Integer.MIN_VALUE || number > Integer.MAX_VALUE) throw invalid("integer overflows field");
+ } else if (type == Short.class || type == short.class) {
+ if (number < Short.MIN_VALUE || number > Short.MAX_VALUE) throw invalid("integer overflows field");
+ } else if (type == Byte.class || type == byte.class) {
+ if (number < Byte.MIN_VALUE || number > Byte.MAX_VALUE) throw invalid("integer overflows field");
+ }
+ return number;
+ }
+ if (type == BigDecimal.class || type == java.math.BigInteger.class) {
+ if (!value.isTextual()) throw invalid("Decimal field has wrong type");
+ try {
+ if (type == java.math.BigInteger.class) {
+ if (!value.textValue().matches("-?(0|[1-9][0-9]*)")
+ || value.textValue().equals("-0")) {
+ throw invalid("integer field has wrong type");
+ }
+ return new java.math.BigInteger(value.textValue()).toString();
+ }
+ BigDecimal decimal = new BigDecimal(value.textValue());
+ if (!fitsDeclaredDecimal(property, decimal)) {
+ throw invalid("Decimal lacks or exceeds modeled precision/scale");
+ }
+ return decimal.toPlainString();
+ }
+ catch (NumberFormatException failure) { throw invalid("Decimal field is invalid"); }
+ }
+ if (type == Double.class || type == double.class || type == Float.class || type == float.class) {
+ if (!value.isNumber() || !Double.isFinite(value.doubleValue())) {
+ throw invalid("floating field has wrong type");
+ }
+ return value.doubleValue();
+ }
+ if (type == byte[].class && value.isObject() && value.size() == 1
+ && value.path("$binary_base64").isTextual()) {
+ try { return java.util.Base64.getDecoder().decode(value.path("$binary_base64").textValue()); }
+ catch (IllegalArgumentException failure) { throw invalid("binary field is invalid"); }
+ }
+ throw invalid("unsupported modeled field type");
+ }
+
+ static void applyOne(SqlExecutionAdapter sql, Mutation mutation) {
+ String table = quoted(mutation.table());
+ String idColumn = quoted(mutation.idColumn());
+ String versionColumn = quoted(mutation.versionColumn());
+ List> current = sql.queryForList("SELECT " + versionColumn
+ + " AS committed_version FROM " + table + " WHERE " + idColumn + " = ?",
+ new Object[] {mutation.id()});
+ Long actualVersion = current.isEmpty() ? null : ((Number) current.get(0).get("committed_version")).longValue();
+ if (!Objects.equals(actualVersion, mutation.baseVersion())) {
+ throw new SqliteCommittedEventStore.CommittedEventException("VERSION_CONFLICT",
+ "committed-event predecessor version mismatch");
+ }
+ List