feat: add PDS data repair script (#208)

Rebuilds PDS records from local DB data. The old code wrote records
with wrong field names (skillName, companyName, startDate, endDate)
which the PDS silently dropped, resulting in incomplete records. The
local DB has the correct data from write-through indexing.

Compares each PDS record against local DB and only rewrites records
that are missing data. Supports --dry-run.

scripts/repair-pds-records.ts

···
+/**
+ * PDS Data Repair Script
+ *
+ * Rebuilds PDS records from local DB data. The old code wrote records with
+ * wrong field names (skillName, companyName, startDate, endDate) which the
+ * PDS silently dropped, resulting in incomplete records. The local DB has
+ * the correct data from write-through indexing.
+ *
+ * This script reads each user's skills, positions, and education from the
+ * local DB and overwrites the PDS records with complete, lexicon-conformant data.
+ *
+ * Usage:
+ *   DATABASE_URL=... tsx scripts/repair-pds-records.ts [--dry-run]
+ *
+ * Delete this script after successful production run.
+ */
+import pg from 'pg';
+import { drizzle } from 'drizzle-orm/node-postgres';
+import { eq } from 'drizzle-orm';
+import Redis from 'ioredis';
+import { Agent } from '@atproto/api';
+import { createOAuthClient } from '../src/oauth/client.js';
+import { profiles } from '../src/db/schema/profiles.js';
+import { skills } from '../src/db/schema/skills.js';
+import { positions } from '../src/db/schema/positions.js';
+import { education } from '../src/db/schema/education.js';
+import { skillPositionLinks } from '../src/db/schema/skill-position-links.js';
+import { loadConfig } from '../src/config.js';
+import { logger } from '../src/logger.js';
+
+const dryRun = process.argv.includes('--dry-run');
+const log = logger.child({ script: 'repair-pds-records', dryRun });
+log.info(dryRun ? 'DRY RUN' : 'LIVE RUN');
+
+const config = loadConfig();
+const pool = new pg.Pool({ connectionString: config.DATABASE_URL });
+const db = drizzle(pool);
+const valkey = new Redis(config.VALKEY_URL ?? 'redis://localhost:6379');
+const oauthClient = await createOAuthClient(config, db, valkey);
+
+// --- Build lexicon-conformant PDS records from local DB rows ---
+
+interface SkillRecord {
+  $type: 'id.sifa.profile.skill';
+  name: string;
+  category?: string;
+  createdAt: string;
+}
+
+interface PositionRecord {
+  $type: 'id.sifa.profile.position';
+  company: string;
+  companyDid?: string;
+  title: string;
+  description?: string;
+  employmentType?: string;
+  workplaceType?: string;
+  location?: { country: string; region?: string; city?: string; countryCode?: string };
+  startedAt: string;
+  endedAt?: string;
+  isPrimary?: boolean;
+  skills?: Array<{ uri: string; cid: string }>;
+  createdAt: string;
+}
+
+interface EducationRecord {
+  $type: 'id.sifa.profile.education';
+  institution: string;
+  institutionDid?: string;
+  degree?: string;
+  fieldOfStudy?: string;
+  description?: string;
+  startedAt?: string;
+  endedAt?: string;
+  createdAt: string;
+}
+
+function buildSkillRecord(row: typeof skills.$inferSelect): SkillRecord {
+  const rec: SkillRecord = {
+    $type: 'id.sifa.profile.skill',
+    name: row.name,
+    createdAt: row.createdAt.toISOString(),
+  };
+  if (row.category) rec.category = row.category;
+  return rec;
+}
+
+function buildPositionRecord(
+  row: typeof positions.$inferSelect,
+  skillRkeys: string[],
+): PositionRecord {
+  const rec: PositionRecord = {
+    $type: 'id.sifa.profile.position',
+    company: row.company,
+    title: row.title,
+    startedAt: row.startedAt,
+    createdAt: row.createdAt.toISOString(),
+  };
+  if (row.companyDid) rec.companyDid = row.companyDid;
+  if (row.description) rec.description = row.description;
+  if (row.employmentType) rec.employmentType = row.employmentType;
+  if (row.workplaceType) rec.workplaceType = row.workplaceType;
+  if (row.locationCountry) {
+    rec.location = { country: row.locationCountry };
+    if (row.locationRegion) rec.location.region = row.locationRegion;
+    if (row.locationCity) rec.location.city = row.locationCity;
+    if (row.countryCode) rec.location.countryCode = row.countryCode;
+  }
+  if (row.endedAt) rec.endedAt = row.endedAt;
+  if (row.isPrimary) rec.isPrimary = true;
+  if (skillRkeys.length > 0) {
+    rec.skills = skillRkeys.map((sk) => ({
+      uri: `at://${row.did}/id.sifa.profile.skill/${sk}`,
+      cid: '', // CID will be resolved by PDS
+    }));
+  }
+  return rec;
+}
+
+function buildEducationRecord(row: typeof education.$inferSelect): EducationRecord {
+  const rec: EducationRecord = {
+    $type: 'id.sifa.profile.education',
+    institution: row.institution,
+    createdAt: row.createdAt.toISOString(),
+  };
+  if (row.institutionDid) rec.institutionDid = row.institutionDid;
+  if (row.degree) rec.degree = row.degree;
+  if (row.fieldOfStudy) rec.fieldOfStudy = row.fieldOfStudy;
+  if (row.description) rec.description = row.description;
+  if (row.startedAt) rec.startedAt = row.startedAt;
+  if (row.endedAt) rec.endedAt = row.endedAt;
+  return rec;
+}
+
+// --- Comparison: check if PDS record matches local DB ---
+
+function recordsMatch(
+  pdsValue: Record<string, unknown>,
+  dbRecord: Record<string, unknown>,
+): boolean {
+  // Compare all fields from the DB record against PDS
+  for (const [key, val] of Object.entries(dbRecord)) {
+    if (key === '$type') continue;
+    const pdsVal = pdsValue[key];
+    if (val === undefined || val === null) continue;
+    if (typeof val === 'object') {
+      // Deep compare for location and skills
+      if (JSON.stringify(val) !== JSON.stringify(pdsVal)) return false;
+    } else if (pdsVal !== val) {
+      return false;
+    }
+  }
+  return true;
+}
+
+// --- Main ---
+
+const allProfiles = await db.select({ did: profiles.did, handle: profiles.handle }).from(profiles);
+log.info({ count: allProfiles.length }, 'Found profiles to repair');
+
+const results = { repaired: 0, alreadyCorrect: 0, sessionExpired: 0, errors: 0, usersProcessed: 0 };
+
+for (const { did, handle } of allProfiles) {
+  try {
+    const session = await oauthClient.restore(did);
+    const agent = new Agent(session);
+    let userRepairs = 0;
+
+    // --- Skills ---
+    const dbSkills = await db.select().from(skills).where(eq(skills.did, did));
+    if (dbSkills.length > 0) {
+      const pdsRes = await agent.com.atproto.repo.listRecords({
+        repo: did,
+        collection: 'id.sifa.profile.skill',
+        limit: 200,
+      });
+      const pdsMap = new Map(
+        pdsRes.data.records.map((r) => [
+          r.uri.split('/').pop()!,
+          r.value as Record<string, unknown>,
+        ]),
+      );
+
+      for (const row of dbSkills) {
+        const dbRecord = buildSkillRecord(row) as unknown as Record<string, unknown>;
+        const pdsValue = pdsMap.get(row.rkey);
+
+        if (!pdsValue || !recordsMatch(pdsValue, dbRecord)) {
+          log.info(
+            {
+              did,
+              handle,
+              collection: 'skill',
+              rkey: row.rkey,
+              name: row.name,
+              hadPds: !!pdsValue,
+            },
+            dryRun ? 'WOULD repair' : 'Repairing',
+          );
+          if (!dryRun) {
+            const writes: Array<Record<string, unknown>> = [];
+            if (pdsValue) {
+              writes.push({
+                $type: 'com.atproto.repo.applyWrites#delete',
+                collection: 'id.sifa.profile.skill',
+                rkey: row.rkey,
+              });
+            }
+            writes.push({
+              $type: 'com.atproto.repo.applyWrites#create',
+              collection: 'id.sifa.profile.skill',
+              rkey: row.rkey,
+              value: dbRecord,
+            });
+            await agent.com.atproto.repo.applyWrites({ repo: did, writes });
+          }
+          userRepairs++;
+        }
+      }
+    }
+
+    // --- Positions ---
+    const dbPositions = await db.select().from(positions).where(eq(positions.did, did));
+    if (dbPositions.length > 0) {
+      const pdsRes = await agent.com.atproto.repo.listRecords({
+        repo: did,
+        collection: 'id.sifa.profile.position',
+        limit: 200,
+      });
+      const pdsMap = new Map(
+        pdsRes.data.records.map((r) => [
+          r.uri.split('/').pop()!,
+          r.value as Record<string, unknown>,
+        ]),
+      );
+
+      // Fetch skill-position links for this user
+      const allLinks = await db
+        .select()
+        .from(skillPositionLinks)
+        .where(eq(skillPositionLinks.did, did));
+      const linksByPosition = new Map<string, string[]>();
+      for (const link of allLinks) {
+        const arr = linksByPosition.get(link.positionRkey) ?? [];
+        arr.push(link.skillRkey);
+        linksByPosition.set(link.positionRkey, arr);
+      }
+
+      for (const row of dbPositions) {
+        const skillRkeys = linksByPosition.get(row.rkey) ?? [];
+        const dbRecord = buildPositionRecord(row, skillRkeys) as unknown as Record<string, unknown>;
+        const pdsValue = pdsMap.get(row.rkey);
+
+        if (!pdsValue || !recordsMatch(pdsValue, dbRecord)) {
+          log.info(
+            {
+              did,
+              handle,
+              collection: 'position',
+              rkey: row.rkey,
+              company: row.company,
+              hadPds: !!pdsValue,
+            },
+            dryRun ? 'WOULD repair' : 'Repairing',
+          );
+          if (!dryRun) {
+            const writes: Array<Record<string, unknown>> = [];
+            if (pdsValue) {
+              writes.push({
+                $type: 'com.atproto.repo.applyWrites#delete',
+                collection: 'id.sifa.profile.position',
+                rkey: row.rkey,
+              });
+            }
+            writes.push({
+              $type: 'com.atproto.repo.applyWrites#create',
+              collection: 'id.sifa.profile.position',
+              rkey: row.rkey,
+              value: dbRecord,
+            });
+            await agent.com.atproto.repo.applyWrites({ repo: did, writes });
+          }
+          userRepairs++;
+        }
+      }
+    }
+
+    // --- Education ---
+    const dbEducation = await db.select().from(education).where(eq(education.did, did));
+    if (dbEducation.length > 0) {
+      const pdsRes = await agent.com.atproto.repo.listRecords({
+        repo: did,
+        collection: 'id.sifa.profile.education',
+        limit: 200,
+      });
+      const pdsMap = new Map(
+        pdsRes.data.records.map((r) => [
+          r.uri.split('/').pop()!,
+          r.value as Record<string, unknown>,
+        ]),
+      );
+
+      for (const row of dbEducation) {
+        const dbRecord = buildEducationRecord(row) as unknown as Record<string, unknown>;
+        const pdsValue = pdsMap.get(row.rkey);
+
+        if (!pdsValue || !recordsMatch(pdsValue, dbRecord)) {
+          log.info(
+            {
+              did,
+              handle,
+              collection: 'education',
+              rkey: row.rkey,
+              institution: row.institution,
+              hadPds: !!pdsValue,
+            },
+            dryRun ? 'WOULD repair' : 'Repairing',
+          );
+          if (!dryRun) {
+            const writes: Array<Record<string, unknown>> = [];
+            if (pdsValue) {
+              writes.push({
+                $type: 'com.atproto.repo.applyWrites#delete',
+                collection: 'id.sifa.profile.education',
+                rkey: row.rkey,
+              });
+            }
+            writes.push({
+              $type: 'com.atproto.repo.applyWrites#create',
+              collection: 'id.sifa.profile.education',
+              rkey: row.rkey,
+              value: dbRecord,
+            });
+            await agent.com.atproto.repo.applyWrites({ repo: did, writes });
+          }
+          userRepairs++;
+        }
+      }
+    }
+
+    if (userRepairs > 0) {
+      results.repaired += userRepairs;
+      log.info({ did, handle, repairs: userRepairs }, 'User repaired');
+    } else {
+      results.alreadyCorrect++;
+    }
+    results.usersProcessed++;
+
+    // Rate limit: 100ms between users
+    await new Promise((r) => setTimeout(r, 100));
+  } catch (err: unknown) {
+    const msg = err instanceof Error ? err.message : String(err);
+    if (
+      msg.includes('invalid_grant') ||
+      msg.includes('expired') ||
+      msg.includes('token_revoked') ||
+      msg.includes('deleted by another process') ||
+      msg.includes('XRPC Not Found')
+    ) {
+      results.sessionExpired++;
+    } else {
+      log.error({ did, handle, err: msg }, 'Failed to repair');
+      results.errors++;
+    }
+  }
+}
+
+log.info(results, 'PDS repair complete');
+valkey.disconnect();
+await pool.end();