feat(rw-backend): site index rebuild (catalog/S3 → DB) via scan + worker queue

.yarnrc.yml

@@ -1,11 +1,11 @@
 nodeLinker: node-modules
 
-# @rwdocs/{core,viewer} 0.1.27 was just published; preapprove it past yarn's
+# @rwdocs/{core,viewer} 0.1.28 was just published; preapprove it past yarn's
 # npmMinimalAgeGate (24h supply-chain quarantine). These are first-party
-# packages we publish. Removable once 0.1.27 is >24h old.
+# packages we publish. Removable once 0.1.28 is >24h old.
 npmPreapprovedPackages:
-  - "@rwdocs/core@^0.1.27"
-  - "@rwdocs/core-darwin-arm64@^0.1.27"
-  - "@rwdocs/core-linux-x64-gnu@^0.1.27"
-  - "@rwdocs/core-linux-x64-musl@^0.1.27"
-  - "@rwdocs/viewer@^0.1.27"
+  - "@rwdocs/core@^0.1.28"
+  - "@rwdocs/core-darwin-arm64@^0.1.28"
+  - "@rwdocs/core-linux-x64-gnu@^0.1.28"
+  - "@rwdocs/core-linux-x64-musl@^0.1.28"
+  - "@rwdocs/viewer@^0.1.28"

plugins/rw-backend/migrations/20260624000000_create_site_index_tables.js

@@ -0,0 +1,42 @@
+// @ts-check
+/** @param {import('knex').Knex} knex */
+exports.up = async function up(knex) {
+  await knex.schema.createTable('section_ownership', table => {
+    table.text('site_ref').notNullable();
+    table.text('section_ref').notNullable();
+    table.text('entity_ref').notNullable();
+    table.text('entity_owner_ref').nullable();
+    table.primary(['site_ref', 'section_ref']);
+    table.index(['entity_owner_ref'], 'section_ownership_owner_idx');
+  });
+  await knex.schema.createTable('sections', table => {
+    table.text('site_ref').notNullable();
+    table.text('section_ref').notNullable();
+    table.text('section_path').notNullable();
+    table.text('parent_section_ref').nullable();
+    table.primary(['site_ref', 'section_ref']);
+  });
+  await knex.schema.createTable('pages', table => {
+    table.text('site_ref').notNullable();
+    table.text('section_ref').notNullable();
+    table.text('subpath').notNullable();
+    table.text('title').notNullable();
+    table.primary(['site_ref', 'section_ref', 'subpath']);
+  });
+  await knex.schema.createTable('site_refresh', table => {
+    table.text('site_ref').primary();
+    table.dateTime('next_update_at').notNullable();
+    table.dateTime('last_built_at').nullable();
+    table.text('result_hash').nullable();
+    table.text('errors').nullable();
+    table.dateTime('last_discovery_at').notNullable();
+    table.index(['next_update_at'], 'site_refresh_next_update_idx');
+  });
+};
+/** @param {import('knex').Knex} knex */
+exports.down = async function down(knex) {
+  await knex.schema.dropTableIfExists('site_refresh');
+  await knex.schema.dropTableIfExists('pages');
+  await knex.schema.dropTableIfExists('sections');
+  await knex.schema.dropTableIfExists('section_ownership');
+};

plugins/rw-backend/package.json

@@ -53,10 +53,11 @@
     "@backstage/plugin-permission-node": "^0.11.1",
     "@backstage/types": "^1.2.2",
     "@rwdocs/backstage-plugin-rw-common": "workspace:^",
-    "@rwdocs/core": "^0.1.27",
+    "@rwdocs/core": "^0.1.28",
     "express": "^4.21.0",
     "express-promise-router": "^4.1.0",
     "luxon": "^3.7.2",
+    "p-limit": "^3.1.0",
     "uuid": "^14.0.0",
     "zod": "^3.25.76 || ^4.0.0"
   },

plugins/rw-backend/src/plugin.ts

@@ -2,7 +2,14 @@ import {
   coreServices,
   createBackendPlugin,
   resolvePackagePath,
+  readSchedulerServiceTaskScheduleDefinitionFromConfig,
 } from "@backstage/backend-plugin-api";
+import { SectionOwnershipStore } from "./siteIndex/SectionOwnershipStore";
+import { RegistryStore } from "./siteIndex/RegistryStore";
+import { SiteRefreshStore } from "./siteIndex/SiteRefreshStore";
+import { runScan } from "./siteIndex/runScan";
+import { runWorker } from "./siteIndex/runWorker";
+import { makeSiteFactory } from "./siteIndex/schedule";
 import { readDurationFromConfig } from "@backstage/config";
 import {
   readRwSiteConfig,
@@ -88,6 +95,40 @@ export const rwPlugin = createBackendPlugin({
         }
         const store = new CommentStore(client);
 
+        const sectionOwnershipStore = new SectionOwnershipStore(client);
+        const registryStore = new RegistryStore(client);
+        const siteRefreshStore = new SiteRefreshStore(client);
+        const makeSite = makeSiteFactory(siteConfig);
+
+        const scanSchedule = config.has("rw.siteIndex.schedule")
+          ? readSchedulerServiceTaskScheduleDefinitionFromConfig(
+              config.getConfig("rw.siteIndex.schedule"),
+            )
+          : { frequency: { minutes: 15 }, timeout: { minutes: 10 }, initialDelay: { seconds: 30 } };
+        const workerSchedule = config.has("rw.siteIndex.worker")
+          ? readSchedulerServiceTaskScheduleDefinitionFromConfig(
+              config.getConfig("rw.siteIndex.worker"),
+            )
+          : { frequency: { seconds: 30 }, timeout: { minutes: 5 }, initialDelay: { seconds: 10 } };
+
+        await scheduler.scheduleTask({
+          id: "rw-site-index-scan",
+          scope: "global",
+          ...scanSchedule,
+          fn: async () =>
+            runScan({ catalog, auth, logger, siteConfig, sectionOwnershipStore, siteRefreshStore }),
+        });
+        await scheduler.scheduleTask({
+          id: "rw-site-index-worker",
+          scope: "local",
+          ...workerSchedule,
+          fn: async () => runWorker({ logger, siteRefreshStore, registryStore, makeSite }),
+        });
+        logger.info(
+          `Scheduled site index rebuild: scan ${JSON.stringify(scanSchedule.frequency)} (global), ` +
+            `worker ${JSON.stringify(workerSchedule.frequency)} (local)`,
+        );
+
         const commentsEnabled = config.getOptionalBoolean("rw.comments.enabled") ?? true;
 
         permissionsRegistry.addResourceType({

plugins/rw-backend/src/siteIndex/RegistryStore.test.ts

@@ -0,0 +1,60 @@
+import type { Knex } from "knex";
+import { createTestDb } from "./__testUtils__/testDb";
+import { RegistryStore } from "./RegistryStore";
+
+describe("RegistryStore", () => {
+  let knex: Knex;
+  beforeEach(async () => (knex = await createTestDb()));
+  afterEach(async () => knex.destroy());
+
+  it("swapSite replaces sections and pages for the site atomically", async () => {
+    const store = new RegistryStore(knex);
+    await store.swapSite(
+      "component:default/a",
+      [
+        {
+          site_ref: "component:default/a",
+          section_ref: "s1",
+          section_path: "",
+          parent_section_ref: null,
+        },
+      ],
+      [{ site_ref: "component:default/a", section_ref: "s1", subpath: "", title: "Home" }],
+    );
+    // seed site b before second swap on a
+    await store.swapSite(
+      "component:default/b",
+      [
+        {
+          site_ref: "component:default/b",
+          section_ref: "sb1",
+          section_path: "b",
+          parent_section_ref: null,
+        },
+      ],
+      [{ site_ref: "component:default/b", section_ref: "sb1", subpath: "bp", title: "B Home" }],
+    );
+    await store.swapSite(
+      "component:default/a",
+      [
+        {
+          site_ref: "component:default/a",
+          section_ref: "s2",
+          section_path: "x",
+          parent_section_ref: "s1",
+        },
+      ],
+      [{ site_ref: "component:default/a", section_ref: "s2", subpath: "p", title: "P" }],
+    );
+
+    expect(
+      await knex("sections").where({ site_ref: "component:default/a" }).pluck("section_ref"),
+    ).toEqual(["s2"]);
+    expect(
+      await knex("pages").where({ site_ref: "component:default/a" }).pluck("section_ref"),
+    ).toEqual(["s2"]);
+    // site b must be untouched by the second swap on a
+    expect(await knex("sections").where({ site_ref: "component:default/b" })).toHaveLength(1);
+    expect(await knex("pages").where({ site_ref: "component:default/b" })).toHaveLength(1);
+  });
+});

plugins/rw-backend/src/siteIndex/RegistryStore.ts

@@ -0,0 +1,16 @@
+import type { Knex } from "knex";
+import type { SectionRow, PageRow } from "./types";
+
+export class RegistryStore {
+  constructor(private readonly knex: Knex) {}
+
+  /** Replace the site's `sections` and `pages` rows in one transaction. */
+  async swapSite(siteRef: string, sections: SectionRow[], pages: PageRow[]): Promise<void> {
+    await this.knex.transaction(async (tx) => {
+      await tx("sections").where({ site_ref: siteRef }).del();
+      await tx("pages").where({ site_ref: siteRef }).del();
+      if (sections.length) await tx.batchInsert("sections", sections, 500);
+      if (pages.length) await tx.batchInsert("pages", pages, 500);
+    });
+  }
+}

plugins/rw-backend/src/siteIndex/SectionOwnershipStore.test.ts

@@ -0,0 +1,47 @@
+import type { Knex } from "knex";
+import { createTestDb } from "./__testUtils__/testDb";
+import { SectionOwnershipStore } from "./SectionOwnershipStore";
+
+describe("SectionOwnershipStore", () => {
+  let knex: Knex;
+  beforeEach(async () => (knex = await createTestDb()));
+  afterEach(async () => knex.destroy());
+
+  it("swapSite replaces only the given site's links", async () => {
+    const store = new SectionOwnershipStore(knex);
+    await store.swapSite("component:default/a", [
+      {
+        site_ref: "component:default/a",
+        section_ref: "s1",
+        entity_ref: "e1",
+        entity_owner_ref: "g1",
+      },
+    ]);
+    await store.swapSite("component:default/b", [
+      {
+        site_ref: "component:default/b",
+        section_ref: "s2",
+        entity_ref: "e2",
+        entity_owner_ref: null,
+      },
+    ]);
+    // re-swap a with new links
+    await store.swapSite("component:default/a", [
+      {
+        site_ref: "component:default/a",
+        section_ref: "s3",
+        entity_ref: "e3",
+        entity_owner_ref: "g3",
+      },
+    ]);
+
+    const rows = await knex("section_ownership").orderBy(["site_ref", "section_ref"]);
+    expect(rows.map((r) => `${r.site_ref}:${r.section_ref}`)).toEqual([
+      "component:default/a:s3",
+      "component:default/b:s2",
+    ]);
+    const rowA = rows.find((r) => r.site_ref === "component:default/a" && r.section_ref === "s3");
+    expect(rowA?.entity_ref).toBe("e3");
+    expect(rowA?.entity_owner_ref).toBe("g3");
+  });
+});

plugins/rw-backend/src/siteIndex/SectionOwnershipStore.ts

@@ -0,0 +1,19 @@
+import type { Knex } from "knex";
+import type { SectionOwnershipRow } from "./types";
+
+const TABLE = "section_ownership";
+
+export class SectionOwnershipStore {
+  constructor(private readonly knex: Knex) {}
+
+  /** Replace all links for `siteRef`. Pass `executor` to join a per-site transaction. */
+  async swapSite(
+    siteRef: string,
+    links: SectionOwnershipRow[],
+    executor?: Knex | Knex.Transaction,
+  ): Promise<void> {
+    const exec = executor ?? this.knex;
+    await exec(TABLE).where({ site_ref: siteRef }).del();
+    if (links.length) await exec.batchInsert(TABLE, links, 500);
+  }
+}

plugins/rw-backend/src/siteIndex/SiteRefreshStore.test.ts

@@ -0,0 +1,73 @@
+import type { Knex } from "knex";
+import { createTestDb } from "./__testUtils__/testDb";
+import { SiteRefreshStore } from "./SiteRefreshStore";
+
+describe("SiteRefreshStore", () => {
+  let knex: Knex;
+  let store: SiteRefreshStore;
+  beforeEach(async () => {
+    knex = await createTestDb();
+    store = new SiteRefreshStore(knex);
+  });
+  afterEach(async () => knex.destroy());
+
+  it("upsertSite inserts new rows as due now and keeps existing schedule on re-scan", async () => {
+    const t1 = new Date("2026-06-24T00:00:00Z");
+    await store.upsertSite("s", t1);
+    let row = await knex("site_refresh").where({ site_ref: "s" }).first();
+    expect(new Date(row.next_update_at).getTime()).toBe(t1.getTime());
+
+    // simulate a completed build pushing next_update_at far out
+    const future = new Date("2026-06-24T01:00:00Z");
+    await store.completeSuccess("s", "h", future, t1);
+
+    const t2 = new Date("2026-06-24T00:30:00Z");
+    await store.upsertSite("s", t2);
+    row = await knex("site_refresh").where({ site_ref: "s" }).first();
+    // next_update_at preserved (not reset to t2); last_discovery_at advanced
+    expect(new Date(row.next_update_at).getTime()).toBe(future.getTime());
+    expect(new Date(row.last_discovery_at).getTime()).toBe(t2.getTime());
+  });
+
+  it("pruneMissing deletes rows not seen this scan", async () => {
+    await store.upsertSite("old", new Date("2026-06-24T00:00:00Z"));
+    await store.upsertSite("new", new Date("2026-06-24T02:00:00Z"));
+    const deleted = await store.pruneMissing(new Date("2026-06-24T01:00:00Z"));
+    expect(deleted).toBe(1);
+    expect(await knex("site_refresh").pluck("site_ref")).toEqual(["new"]);
+  });
+
+  it("claimDue returns due rows oldest-first and bumps next_update_at to the lease", async () => {
+    await store.upsertSite("a", new Date("2026-06-24T00:00:00Z"));
+    await store.upsertSite("b", new Date("2026-06-24T00:00:01Z"));
+    const lease = new Date("2026-06-24T03:00:00Z");
+    const claimed = await store.claimDue(new Date("2026-06-24T01:00:00Z"), 10, lease);
+    expect(claimed.map((c) => c.siteRef)).toEqual(["a", "b"]);
+    // both leased out → no longer due before lease
+    const again = await store.claimDue(new Date("2026-06-24T02:00:00Z"), 10, lease);
+    expect(again).toEqual([]);
+  });
+
+  it("completeSuccess sets last_built_at/result_hash/next_update_at and clears errors", async () => {
+    await store.upsertSite("a", new Date("2026-06-24T00:00:00Z"));
+    await store.recordError("a", "boom");
+    const next = new Date("2026-06-24T04:00:00Z");
+    const now = new Date("2026-06-24T00:05:00Z");
+    await store.completeSuccess("a", "hash1", next, now);
+    const row = await knex("site_refresh").where({ site_ref: "a" }).first();
+    expect(row.result_hash).toBe("hash1");
+    expect(row.errors).toBeNull();
+    expect(new Date(row.last_built_at).getTime()).toBe(now.getTime());
+    expect(new Date(row.next_update_at).getTime()).toBe(next.getTime());
+  });
+
+  it("allBuilt is false until every row has last_built_at", async () => {
+    await store.upsertSite("a", new Date("2026-06-24T00:00:00Z"));
+    await store.upsertSite("b", new Date("2026-06-24T00:00:00Z"));
+    expect(await store.allBuilt()).toBe(false);
+    await store.completeSuccess("a", "h", new Date(), new Date());
+    expect(await store.allBuilt()).toBe(false);
+    await store.completeSuccess("b", "h", new Date(), new Date());
+    expect(await store.allBuilt()).toBe(true);
+  });
+});