feat(db): Option B schema — ContentCurrent + ContentHistory (Workstream D, draft)

db-qa/schema.cds

@@ -7,6 +7,11 @@ entity ContentFiles : shared.ContentFilesAspect {}
 
 entity ContentManifest : shared.ContentManifestAspect {}
 
+// Option B (slug-targeted-delta-rebuild) — QA-channel parity with prod.
+entity ContentCurrent : shared.ContentCurrentAspect {}
+
+entity ContentHistory : shared.ContentHistoryAspect {}
+
 // Plain-text projection of published Hugo HTML, indexed for full-text search.
 // Replaced (not versioned) on every publish so search reflects current content.
 @cds.autoexpose: false

db/_content-shape.cds

@@ -64,6 +64,41 @@ aspect ContentManifestAspect : managed {
   firstAppendAt             : Timestamp;
 }
 
+// Option B (slug-targeted-delta-rebuild): the MUTABLE current-content table —
+// one row per slug, NO version column. Readers hit this directly (WHERE slug=?)
+// instead of joining on the active manifest version, so a publish writes ONLY
+// the changed slugs (no O(corpus) carry-forward). `sourceVersion` records the
+// manifest version that last wrote this slug (audit + cache-generation token).
+aspect ContentCurrentAspect : managed {
+  key slug                  : String(255);
+  content                   : LargeBinary;
+  contentHash               : Sha256;
+  sizeBytes                 : Integer;
+  compressedBytes           : Integer;
+  mimeType                  : String(100) default 'text/html';
+  sourceContent             : LargeBinary;
+  sourceHash                : Sha256;
+  sourceVersion             : Integer;
+}
+
+// Option B: append-only per-(version, slug) history for drift-detection
+// (detectReverts) + rollback replay. Carries the BLOB (`content`/`sourceContent`)
+// so rollback is a self-contained replay into ContentCurrent (design.md D2);
+// GC'd by the repurposed cleanupContentVersions. `action=DELETED` tombstones a
+// slug removed at that version so rollback can re-add/remove correctly.
+aspect ContentHistoryAspect : managed {
+  key version               : Integer;
+  key slug                  : String(255);
+  action                    : String(10) enum { WRITTEN; DELETED; };
+  content                   : LargeBinary;
+  contentHash               : Sha256;
+  sizeBytes                 : Integer;
+  compressedBytes           : Integer;
+  mimeType                  : String(100) default 'text/html';
+  sourceContent             : LargeBinary;
+  sourceHash                : Sha256;
+}
+
 aspect TutorialBodyTextAspect : managed {
   key slug                  : String(255);
   bodyText                  : LargeString;

db/schema.cds

@@ -598,6 +598,13 @@ entity ContentFiles : shared.ContentFilesAspect {}
 
 entity ContentManifest : shared.ContentManifestAspect {}
 
+// Option B (slug-targeted-delta-rebuild): mutable current-content table +
+// append-only history. Coexist with ContentFiles/ContentManifest during the
+// flag-gated dual-write migration; ContentFiles is retired one release after cutover.
+entity ContentCurrent : shared.ContentCurrentAspect {}
+
+entity ContentHistory : shared.ContentHistoryAspect {}
+
 // Plain-text projection of published Hugo HTML, indexed for full-text search.
 // Replaced (not versioned) on every publish so search reflects current content.
 @cds.autoexpose: false

srv/lib/content-publish-session.js

@@ -461,6 +461,21 @@ export function createSessionHelpers({ namespace }) {
     // srv/lib/content-store.js:320-378 so prod/SQLite parity is preserved.
     const { carriedForward, carriedSize } = await carryForwardUnchanged(namespace, newVersion, hanaTableName, getActiveVersion);
 
+    // Option B dual-write (Workstream D, flag-gated, fail-safe). Mirror the
+    // freshly-published slugs into the mutable ContentCurrent + append-only
+    // ContentHistory alongside the legacy ContentFiles write, so readers can be
+    // cut over behind a separate read flag with a safe fallback. Never throws
+    // into the commit tx — the legacy write remains the source of truth until
+    // the read cutover.
+    if (process.env.CONTENT_DELTA_WRITE_ENABLED === 'true') {
+      try {
+        const { written } = await dualWriteCurrentAndHistory(namespace, newVersion, freshSlugs, hanaTableName);
+        LOG.info(`[content/publish/commit] Option B dual-write: ${written} slug(s) → ContentCurrent/ContentHistory`);
+      } catch (err) {
+        LOG.error('[content/publish/commit] Option B dual-write failed (non-fatal; legacy ContentFiles write is source of truth):', err.message);
+      }
+    }
+
     // Compute aggregated size after carry-forward for the manifest stats.
     const freshAgg = await SELECT.one.from(ContentFiles)
       .columns('count(*) as c', 'sum(sizeBytes) as s')
@@ -1303,7 +1318,93 @@ async function carryForwardUnchanged(namespace, newVersion, hanaTableName, getAc
 }
 
 // ---------------------------------------------------------------------------
-// Recompute TUTORIAL TaskRecords progress for any tutorial whose body content
+// Option B dual-write (Workstream D, slug-targeted-delta-rebuild). When the
+// CONTENT_DELTA_WRITE_ENABLED flag is on, mirror the freshly-published slugs
+// into the mutable ContentCurrent table (UPSERT — one row per slug, no version)
+// and append a WRITTEN row per (version, slug) to the append-only ContentHistory.
+// This runs ALONGSIDE the legacy ContentFiles write during the migration window
+// (dual-write), so readers can be cut over behind a separate read flag with a
+// safe rollback to ContentFiles. Fail-SAFE: any fault here is logged and
+// swallowed — it must never break the legacy commit (which remains the source
+// of truth until the read cutover).
+//
+// BLOB handling mirrors carryForwardUnchanged: chunked, raw db.run on HANA to
+// materialize LOBs as buffers (LOB-locator gotcha), CQL on SQLite.
+async function dualWriteCurrentAndHistory(namespace, newVersion, freshSlugs, hanaTableName) {
+  if (!freshSlugs || freshSlugs.length === 0) return { written: 0 };
+  const ents = cds.entities(namespace);
+  const { ContentFiles, ContentCurrent, ContentHistory } = ents;
+  if (!ContentCurrent || !ContentHistory) {
+    LOG.warn('[content/publish/commit] dual-write skipped — ContentCurrent/ContentHistory not in model');
+    return { written: 0 };
+  }
+
+  const db = await cds.connect.to('db');
+  const isHana = db.options?.kind === 'hana' || db.constructor?.name === 'HANAService';
+  const CHUNK = 50;
+  let written = 0;
+
+  for (let i = 0; i < freshSlugs.length; i += CHUNK) {
+    const chunk = freshSlugs.slice(i, i + CHUNK);
+
+    let rows;
+    if (isHana) {
+      const placeholders = chunk.map(() => '?').join(', ');
+      const raw = await db.run(
+        `SELECT "SLUG", "CONTENT", "CONTENTHASH", "SIZEBYTES", "COMPRESSEDBYTES", "MIMETYPE", "SOURCECONTENT", "SOURCEHASH"
+           FROM "${hanaTableName()}"
+          WHERE "VERSION" = ? AND "SLUG" IN (${placeholders})`,
+        [newVersion, ...chunk]
+      );
+      rows = raw.map((r) => ({
+        slug: r.SLUG, content: r.CONTENT, contentHash: r.CONTENTHASH,
+        sizeBytes: r.SIZEBYTES, compressedBytes: r.COMPRESSEDBYTES,
+        mimeType: r.MIMETYPE, sourceContent: r.SOURCECONTENT, sourceHash: r.SOURCEHASH,
+      }));
+    } else {
+      rows = await SELECT.from(ContentFiles)
+        .columns('slug', 'content', 'contentHash', 'sizeBytes', 'compressedBytes', 'mimeType', 'sourceContent', 'sourceHash')
+        .where({ version: newVersion, slug: { in: chunk } });
+    }
+
+    const currentEntries = [];
+    const historyEntries = [];
+    for (const row of rows) {
+      const buf = Buffer.isBuffer(row.content) ? row.content : await toBuffer(row.content);
+      let srcBuf = null;
+      if (row.sourceContent != null) {
+        srcBuf = Buffer.isBuffer(row.sourceContent) ? row.sourceContent : await toBuffer(row.sourceContent);
+      }
+      currentEntries.push({
+        slug: row.slug, content: buf, contentHash: row.contentHash,
+        sizeBytes: row.sizeBytes, compressedBytes: row.compressedBytes,
+        mimeType: row.mimeType, sourceContent: srcBuf, sourceHash: row.sourceHash ?? null,
+        sourceVersion: newVersion,
+      });
+      historyEntries.push({
+        version: newVersion, slug: row.slug, action: 'WRITTEN', content: buf,
+        contentHash: row.contentHash, sizeBytes: row.sizeBytes, compressedBytes: row.compressedBytes,
+        mimeType: row.mimeType, sourceContent: srcBuf, sourceHash: row.sourceHash ?? null,
+      });
+    }
+
+    // UPSERT ContentCurrent by replace (DELETE-then-INSERT keyed on slug) —
+    // portable across SQLite + HANA and avoids relying on native UPSERT.
+    const chunkSlugs = currentEntries.map((e) => e.slug);
+    if (chunkSlugs.length) {
+      await DELETE.from(ContentCurrent).where({ slug: { in: chunkSlugs } });
+      await INSERT.into(ContentCurrent).entries(currentEntries);
+      // History is append-only keyed on (version, slug); a re-commit of the same
+      // version (idempotent retry) would duplicate-key, so clear this version's
+      // rows for the chunk first.
+      await DELETE.from(ContentHistory).where({ version: newVersion, slug: { in: chunkSlugs } });
+      await INSERT.into(ContentHistory).entries(historyEntries);
+      written += currentEntries.length;
+    }
+  }
+
+  return { written };
+}
 // was published in this version. appendToSession already calls the bulk
 // recompute when metadata is provided, but if a chunk arrived with body text
 // only (no metadata payload), the recompute would be skipped. Re-running here

test/unit/content-delta-dualwrite.test.js

@@ -0,0 +1,118 @@
+// test/unit/content-delta-dualwrite.test.js
+//
+// Workstream D (slug-targeted-delta-rebuild) — Option B dual-write guard.
+//
+// When CONTENT_DELTA_WRITE_ENABLED=true, commitSession mirrors the freshly-
+// published slugs into the mutable ContentCurrent table (one row per slug, no
+// version) + appends WRITTEN rows to ContentHistory, ALONGSIDE the legacy
+// ContentFiles write. This test drives publishes on in-memory SQLite and
+// asserts: (a) ContentCurrent is one-row-per-slug and UPSERTs on republish,
+// (b) ContentHistory accumulates per version, (c) the flag OFF writes neither.
+//
+// HANA LOB-locator behavior is NOT exercised here (SQLite CQL path); that is
+// covered by the hybrid publish→rollback test in Workstream D task 7.4.
+
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import cds from '@sap/cds';
+import { gzipSync } from 'node:zlib';
+import { createSessionHelpers } from '../../srv/lib/content-publish-session.js';
+
+const NS = 'com.sap.developers.ims';
+
+cds.test('serve', '--project', '.', '--in-memory');
+
+function html(s) {
+  return gzipSync(Buffer.from(`<html><body><main class="tutorial-main">${s}</main></body></html>`, 'utf-8')).toString('base64');
+}
+function source(s) {
+  return gzipSync(Buffer.from(s, 'utf-8')).toString('base64');
+}
+async function appendAll(helpers, sessionId, slugs) {
+  const files = {};
+  const sources = {};
+  for (const slug of slugs) { files[slug] = html(`body-${slug}`); sources[slug] = source(`src-${slug}`); }
+  await helpers.appendToSession({ sessionId, files, sources });
+}
+
+describe('Option B dual-write (Workstream D)', () => {
+  let helpers;
+  let ContentFiles, ContentManifest, ContentCurrent, ContentHistory, PipelineLog, JobLocks;
+  const prevFlag = process.env.CONTENT_DELTA_WRITE_ENABLED;
+
+  beforeAll(() => {
+    helpers = createSessionHelpers({ namespace: NS });
+    ({ ContentFiles, ContentManifest, ContentCurrent, ContentHistory, PipelineLog, JobLocks } = cds.entities(NS));
+  });
+  afterAll(() => {
+    if (prevFlag === undefined) delete process.env.CONTENT_DELTA_WRITE_ENABLED;
+    else process.env.CONTENT_DELTA_WRITE_ENABLED = prevFlag;
+  });
+  beforeEach(async () => {
+    await DELETE.from(ContentFiles);
+    await DELETE.from(ContentManifest);
+    await DELETE.from(ContentCurrent);
+    await DELETE.from(ContentHistory);
+    await DELETE.from(PipelineLog);
+    await DELETE.from(JobLocks);
+  });
+
+  it('exposes ContentCurrent + ContentHistory entities', () => {
+    expect(ContentCurrent).toBeTruthy();
+    expect(ContentHistory).toBeTruthy();
+  });
+
+  it('writes ContentCurrent (one row per slug) + ContentHistory when the flag is ON', async () => {
+    process.env.CONTENT_DELTA_WRITE_ENABLED = 'true';
+    const slugs = ['a', 'b', 'c'];
+    const s = await helpers.beginPublishSession({ trigger: 'ci/test', expectedSlugCount: slugs.length, initiator: 'test' });
+    await appendAll(helpers, s.sessionId, slugs);
+    const res = await helpers.commitSession({ sessionId: s.sessionId });
+
+    const current = await SELECT.from(ContentCurrent).columns('slug', 'contentHash', 'sourceVersion', 'content');
+    expect(current.map(r => r.slug).sort()).toEqual(['a', 'b', 'c']);
+    for (const row of current) {
+      expect(row.content, `ContentCurrent.${row.slug} has null content`).toBeTruthy();
+      expect(row.sourceVersion).toBe(res.version);
+    }
+    const history = await SELECT.from(ContentHistory).columns('slug', 'version', 'action');
+    expect(history.length).toBe(3);
+    expect(history.every(h => h.action === 'WRITTEN' && h.version === res.version)).toBe(true);
+  }, 60_000);
+
+  it('UPSERTs ContentCurrent on republish (stays one row per slug) + appends history per version', async () => {
+    process.env.CONTENT_DELTA_WRITE_ENABLED = 'true';
+    const slugs = ['a', 'b', 'c'];
+    const s1 = await helpers.beginPublishSession({ trigger: 'ci/test', expectedSlugCount: 3, initiator: 'test' });
+    await appendAll(helpers, s1.sessionId, slugs);
+    const r1 = await helpers.commitSession({ sessionId: s1.sessionId });
+
+    // Republish only 'a' with new content.
+    const s2 = await helpers.beginPublishSession({ trigger: 'ci/test', expectedSlugCount: 1, initiator: 'test' });
+    await helpers.appendToSession({ sessionId: s2.sessionId, files: { a: html('body-a-v2') }, sources: { a: source('src-a-v2') } });
+    const r2 = await helpers.commitSession({ sessionId: s2.sessionId });
+
+    // ContentCurrent still has exactly one row for 'a', now at the new version.
+    const aRows = await SELECT.from(ContentCurrent).where({ slug: 'a' });
+    expect(aRows.length).toBe(1);
+    expect(aRows[0].sourceVersion).toBe(r2.version);
+    // b + c unchanged rows remain (from v1) — dual-write only touches fresh slugs.
+    const all = await SELECT.from(ContentCurrent).columns('slug');
+    expect(all.map(r => r.slug).sort()).toEqual(['a', 'b', 'c']);
+
+    // History has 'a' at both versions (append-only).
+    const aHistory = await SELECT.from(ContentHistory).where({ slug: 'a' });
+    expect(aHistory.map(h => h.version).sort((x, y) => x - y)).toEqual([r1.version, r2.version]);
+  }, 60_000);
+
+  it('writes NEITHER table when the flag is OFF', async () => {
+    process.env.CONTENT_DELTA_WRITE_ENABLED = 'false';
+    const s = await helpers.beginPublishSession({ trigger: 'ci/test', expectedSlugCount: 2, initiator: 'test' });
+    await appendAll(helpers, s.sessionId, ['x', 'y']);
+    await helpers.commitSession({ sessionId: s.sessionId });
+
+    expect((await SELECT.from(ContentCurrent)).length).toBe(0);
+    expect((await SELECT.from(ContentHistory)).length).toBe(0);
+    // Legacy ContentFiles still written (source of truth).
+    expect((await SELECT.from(ContentFiles).columns('slug')).map(r => r.slug).sort()).toEqual(['x', 'y']);
+  }, 60_000);
+});