[Feature] Scheduled GitHub source drift-check job (#1176)

.changeset/gh-source-drift-scheduler.md

@@ -0,0 +1,5 @@
+---
+"ornn-api": minor
+---
+
+Add the scheduled GitHub source drift-check job (#1176): a multi-pod-safe Agenda scheduler periodically probes every GitHub-sourced skill's upstream (coalescing skills that share a repo/ref into one request), records whether it has drifted, and notifies the owner when a source repo becomes unreachable. Cadence is driven by the `sourceSync.pollSchedule` setting; the job honors GitHub rate limits and never lets one skill's failure abort the run. It only records drift state — automatic re-publish is a later change.

ornn-api/src/bootstrap.ts

@@ -109,6 +109,11 @@ import {
   createMirrorScheduler,
   type MirrorScheduler,
 } from "./domains/skills/mirror/scheduler";
+import {
+  createSourceSyncScheduler,
+  type SourceSyncScheduler,
+} from "./domains/skills/crud/sourceSyncScheduler";
+import { runSourceDriftJob } from "./domains/skills/crud/sourceDriftJob";
 
 // Domain: Me (caller-scoped endpoints)
 import { createMeRoutes } from "./domains/me/routes";
@@ -784,6 +789,39 @@ export async function bootstrap(
     mirrorScheduler,
   });
 
+  // In-process source-drift scheduler (#1176). Same multi-pod-safe Agenda
+  // pattern as the mirror scheduler; cadence driven by
+  // `settings.sourceSync.pollSchedule`. Detects when a GitHub-sourced skill's
+  // upstream moved and records drift state (no publish — that is #1177). A
+  // start failure logs and leaves this pod without the scheduled scan rather
+  // than crashing boot.
+  let sourceSyncScheduler: SourceSyncScheduler | null = null;
+  try {
+    sourceSyncScheduler = createSourceSyncScheduler({
+      db,
+      logger,
+      settingsService,
+      runDriftJob: () =>
+        runSourceDriftJob({
+          skillRepo,
+          settingsService,
+          envTokenFallback: config.sourceSyncGithubToken,
+          notifier: notificationService,
+          logger,
+          // Small per-group jitter so a large catalogue spreads its probes
+          // across the tick instead of bursting one egress IP.
+          jitterMs: 250,
+        }),
+    });
+    await sourceSyncScheduler.start();
+  } catch (err) {
+    logger.error(
+      { err: err instanceof Error ? err.message : String(err) },
+      "source-sync scheduler failed to start — scheduled drift checks will not run on this pod",
+    );
+    sourceSyncScheduler = null;
+  }
+
   // Skill routes — sharing is now a direct PUT /permissions write; the
   // audit signal is surfaced as a per-version label, not a gate.
   // #1136 — forward reference: the skill routes fire a reactive skillset
@@ -1134,9 +1172,16 @@ export async function bootstrap(
   // ---- Shutdown ----
   async function shutdown(): Promise<void> {
     logger.info("Shutting down ornn-api...");
-    // Stop the scheduler first so no new mirror reconciles start while
-    // we're tearing the Mongo connection down. `stop()` is idempotent +
-    // already swallows its own errors.
+    // Stop the schedulers first so no new reconciles / drift checks start
+    // while we're tearing the Mongo connection down. `stop()` is idempotent +
+    // already swallows its own errors. Source-sync first, then mirror.
+    if (sourceSyncScheduler) {
+      try {
+        await sourceSyncScheduler.stop();
+      } catch (err) {
+        logger.warn({ err }, "Source-sync scheduler stop failed — continuing");
+      }
+    }
     if (mirrorScheduler) {
       try {
         await mirrorScheduler.stop();

ornn-api/src/domains/notifications/service.ts

@@ -442,6 +442,32 @@ export class NotificationService {
     });
   }
 
+  /**
+   * Owner-side notification (#1176) fired when an automatic drift check finds
+   * a GitHub-sourced skill's upstream repo/ref can no longer be resolved
+   * (deleted, made private, or the branch/tag was removed). Lets the owner
+   * re-link or fix the source. Deep-links to the skill detail page.
+   */
+  async notifySourceBroken(params: {
+    ownerId: string;
+    skillGuid: string;
+    repo: string;
+    ref: string;
+  }): Promise<void> {
+    const title = `The GitHub source for one of your skills is unavailable`;
+    const body =
+      `Ornn could not reach the upstream source (${params.repo}@${params.ref}) during an ` +
+      `automatic sync check — it may have been deleted, made private, or the branch/tag ` +
+      `was removed. Re-link the skill to a valid GitHub folder to resume automatic syncing.`;
+    await this.emit(params.ownerId, {
+      category: "skill.source_broken",
+      title,
+      body,
+      link: `/skills/${encodeURIComponent(params.skillGuid)}`,
+      data: { skillGuid: params.skillGuid, repo: params.repo, ref: params.ref },
+    });
+  }
+
   private async emit(
     userId: string,
     payload: {

ornn-api/src/domains/notifications/types.ts

@@ -35,6 +35,9 @@ export const NOTIFICATION_CATEGORIES = [
   "launchPromo.codeDelivered",
   // A member skill became unreadable to the skillset owner (#1136).
   "skillset.member_unreadable",
+  // A GitHub-sourced skill's upstream repo/ref could not be resolved during
+  // an automatic drift check — the source is broken (404/private/deleted) (#1176).
+  "skill.source_broken",
 ] as const;
 
 export type NotificationCategory = (typeof NOTIFICATION_CATEGORIES)[number];

ornn-api/src/domains/skills/crud/repository.sourceDrift.test.ts

@@ -0,0 +1,133 @@
+/**
+ * Repository tests for the source-drift query + setter (#1176), against a
+ * real in-memory MongoDB so the `$not`-regex query, the partial index, and
+ * the dot-path `$set` are exercised for real.
+ *
+ * @module domains/skills/crud/repository.sourceDrift.test
+ */
+import { afterAll, beforeAll, beforeEach, describe, expect, test } from "bun:test";
+import { MongoClient, type Db } from "mongodb";
+import { MongoMemoryServer } from "mongodb-memory-server";
+import { SkillRepository } from "./repository";
+
+let mongo: MongoMemoryServer;
+let client: MongoClient;
+let db: Db;
+let repo: SkillRepository;
+
+beforeAll(async () => {
+  mongo = await MongoMemoryServer.create();
+  client = new MongoClient(mongo.getUri());
+  await client.connect();
+  db = client.db("source_drift_test");
+  repo = new SkillRepository(db);
+  await repo.ensureIndexes();
+});
+
+afterAll(async () => {
+  await client.close();
+  await mongo.stop();
+});
+
+beforeEach(async () => {
+  await db.collection("skills").deleteMany({});
+});
+
+function skillDoc(overrides: Record<string, unknown>): Record<string, unknown> {
+  const now = new Date();
+  return {
+    name: `s-${overrides._id}`,
+    description: "d",
+    metadata: { category: "plain" },
+    skillHash: "h",
+    storageKey: "k",
+    createdBy: "owner",
+    createdOn: now,
+    updatedBy: "owner",
+    updatedOn: now,
+    isPrivate: true,
+    sharedWithUsers: [],
+    sharedWithOrgs: [],
+    latestVersion: "1.0",
+    ...overrides,
+  };
+}
+
+async function seed(...docs: Array<Record<string, unknown>>): Promise<void> {
+  await db.collection("skills").insertMany(docs.map(skillDoc) as never);
+}
+
+describe("findGithubSourcedSkills", () => {
+  test("selects due github skills only; excludes fresh, pinned, and source-less", async () => {
+    const threeHoursAgo = new Date(Date.now() - 3 * 60 * 60 * 1000);
+    await seed(
+      // github, never checked → included
+      { _id: "g1", createdBy: "owner-1", source: { type: "github", repo: "a/x", ref: "main", path: "" } },
+      // github, checked long ago → included
+      {
+        _id: "g2",
+        createdBy: "owner-2",
+        source: { type: "github", repo: "a/x", ref: "main", path: "", lastCheckedAt: threeHoursAgo },
+      },
+      // github, checked just now → excluded (fresh)
+      {
+        _id: "g3",
+        createdBy: "owner-3",
+        source: { type: "github", repo: "b/y", ref: "main", path: "", lastCheckedAt: new Date() },
+      },
+      // pinned 40-hex ref → excluded (never drifts)
+      { _id: "g4", createdBy: "owner-4", source: { type: "github", repo: "c/z", ref: "a".repeat(40), path: "" } },
+      // hand-uploaded (no source) → excluded
+      { _id: "g5", createdBy: "owner-5" },
+    );
+
+    const cutoff = new Date(Date.now() - 60 * 60 * 1000); // 1h ago
+    const due = await repo.findGithubSourcedSkills({ notCheckedSince: cutoff });
+
+    expect(due.map((d) => d.guid).sort()).toEqual(["g1", "g2"]);
+    const g1 = due.find((d) => d.guid === "g1")!;
+    expect(g1.ownerId).toBe("owner-1");
+    expect(g1.source.repo).toBe("a/x");
+    expect(g1.source.ref).toBe("main");
+  });
+});
+
+describe("updateSourceDriftState", () => {
+  test("sets only drift dot-paths; preserves lastSyncedCommit and does NOT bump updatedOn", async () => {
+    const created = new Date("2026-06-01T00:00:00.000Z");
+    await seed({
+      _id: "g1",
+      updatedOn: created,
+      updatedBy: "orig",
+      source: { type: "github", repo: "a/x", ref: "main", path: "", lastSyncedCommit: "keepme" },
+    });
+
+    const checkedAt = new Date("2026-07-01T08:00:00.000Z");
+    await repo.updateSourceDriftState("g1", {
+      driftState: "drifted",
+      upstreamHeadSha: "newsha",
+      etag: 'W/"e1"',
+      lastCheckedAt: checkedAt,
+    });
+
+    const doc = await repo.findByGuid("g1");
+    expect(doc).not.toBeNull();
+    const src = doc!.source!;
+    expect(src.driftState).toBe("drifted");
+    expect(src.upstreamHeadSha).toBe("newsha");
+    expect(src.etag).toBe('W/"e1"');
+    expect(src.lastCheckedAt?.toISOString()).toBe(checkedAt.toISOString());
+    // Load-bearing: the refresh-owned field is untouched, and a background
+    // drift check must not disturb sort-by-updated ordering.
+    expect(src.lastSyncedCommit).toBe("keepme");
+    expect(doc!.updatedOn.toISOString()).toBe(created.toISOString());
+    expect(doc!.updatedBy).toBe("orig");
+  });
+
+  test("is a no-op on a skill with no source", async () => {
+    await seed({ _id: "g1" }); // no source
+    await repo.updateSourceDriftState("g1", { driftState: "broken", lastCheckedAt: new Date() });
+    const doc = await repo.findByGuid("g1");
+    expect(doc!.source).toBeUndefined();
+  });
+});