From 24eb9feebaf62963f9c77f673d13be556f6acd8b Mon Sep 17 00:00:00 2001
From: Florian <45694132+flo-bit@users.noreply.github.com>
Date: Fri, 7 Aug 2026 06:24:25 +0200
Subject: [PATCH] Adopt legacy serving cursors

---
 packages/contrail/src/core/db/records.ts         | 16 +++++++++++++---
 .../tests/serving-source-position.test.ts        | 16 ++++++++++++++++
 2 files changed, 29 insertions(+), 3 deletions(-)

diff --git a/packages/contrail/src/core/db/records.ts b/packages/contrail/src/core/db/records.ts
index 2b34a39..5f8decb 100644
--- a/packages/contrail/src/core/db/records.ts
+++ b/packages/contrail/src/core/db/records.ts
@@ -566,10 +566,20 @@ export async function assertServingSourceCompatibility(
 ): Promise<void> {
   if (!orderedSource) return;
   const existing = await getServingSourcePosition(db);
+  if (!existing) {
+    const legacyCursor = await getLastCursor(db);
+    if (legacyCursor !== null) {
+      await saveOrderedSourcePositionStatement(
+        db,
+        orderedSource,
+        legacyCursor,
+      ).run();
+    }
+    return;
+  }
   if (
-    existing &&
-    (existing.position.source !== orderedSource.source ||
-      existing.position.epoch !== orderedSource.epoch)
+    existing.position.source !== orderedSource.source ||
+    existing.position.epoch !== orderedSource.epoch
   ) {
     throw new Error(
       `configured ordered source ${orderedSource.source}/${orderedSource.epoch} ` +
diff --git a/packages/contrail/tests/serving-source-position.test.ts b/packages/contrail/tests/serving-source-position.test.ts
index 9266732..f6a00da 100644
--- a/packages/contrail/tests/serving-source-position.test.ts
+++ b/packages/contrail/tests/serving-source-position.test.ts
@@ -42,6 +42,22 @@ describe("serving source positions", () => {
     });
   });
 
+  it("adopts an existing legacy cursor when an ordered source is configured", async () => {
+    const db = createSqliteDatabase(":memory:");
+    await initSchema(db, config);
+    await saveCursor(db, 777);
+
+    await new Contrail(config).init(db);
+
+    expect(await getServingSourcePosition(db)).toMatchObject({
+      position: {
+        source: "jetstream",
+        epoch: "primary-2026",
+        cursor: "777",
+      },
+    });
+  });
+
   it("rejects a configured continuity epoch that differs from durable state", async () => {
     const db = createSqliteDatabase(":memory:");
     await initSchema(db, config);
-- 
2.51.2