diff --git a/db-qa/schema.cds b/db-qa/schema.cds index d09056120..906b527d1 100644 --- a/db-qa/schema.cds +++ b/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 diff --git a/db/_content-shape.cds b/db/_content-shape.cds index d0cf714b2..35c9bd076 100644 --- a/db/_content-shape.cds +++ b/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; diff --git a/db/schema.cds b/db/schema.cds index c2be7e051..07ffce7df 100644 --- a/db/schema.cds +++ b/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 diff --git a/srv/lib/content-publish-session.js b/srv/lib/content-publish-session.js index 98d9d3794..638d7b7e2 100644 --- a/srv/lib/content-publish-session.js +++ b/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 diff --git a/test/unit/content-delta-dualwrite.test.js b/test/unit/content-delta-dualwrite.test.js new file mode 100644 index 000000000..b4a61cca1 --- /dev/null +++ b/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(`
${s}
`, '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); +});