fix(api): write-through to local DB after PDS writes (#81)

* fix(api): write-through to local DB after PDS writes

After a successful PDS write, immediately index the record in the local
database so GET requests reflect the change without waiting for the
Jetstream firehose to deliver the event. This eliminates the race
condition where page refresh showed stale data (skills, positions,
education, etc. appearing to vanish after adding/removing).

Extract DB write logic from Jetstream indexers into a shared
record-indexer service. Both write routes and Jetstream handlers now
call the same idempotent upsert functions -- when Jetstream eventually
delivers the duplicate event, the ON CONFLICT clause makes it a no-op.

Follows the write-through pattern used by Cloudflare's Statusphere
(the canonical ATProto AppView example) for read-your-own-writes
consistency on custom lexicons.

Closes #79

* style: format new files with prettier

* fix(tests): use unique canonical skill slug to avoid cross-suite FK conflict

The record-indexer test and skill-normalization-db test both created a
canonical skill with slug 'javascript'. When running together, cleanup
order caused FK violations. Use a unique slug 'writethrough-testlang'
to isolate the test data.

src/jetstream/indexers/certification.ts

···
 import type { Database } from '../../db/index.js';
-import { certifications } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
+import { indexRecord, deleteRecord } from '../../services/record-indexer.js';
+
+const COLLECTION = 'id.sifa.profile.certification';
 
 export function createCertificationIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db
-        .delete(certifications)
-        .where(and(eq(certifications.did, did), eq(certifications.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted certification');
+      await deleteRecord(db, COLLECTION, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(certifications)
-      .values({
-        did,
-        rkey,
-        name: sanitize(record.name as string),
-        authority: sanitizeOptional(record.authority as string | undefined) ?? null,
-        credentialId: (record.credentialId as string) ?? null,
-        credentialUrl: (record.credentialUrl as string) ?? null,
-        issuedAt: (record.issuedAt as string) ?? null,
-        expiresAt: (record.expiresAt as string) ?? null,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [certifications.did, certifications.rkey],
-        set: {
-          name: sanitize(record.name as string),
-          authority: sanitizeOptional(record.authority as string | undefined) ?? null,
-          credentialId: (record.credentialId as string) ?? null,
-          credentialUrl: (record.credentialUrl as string) ?? null,
-          issuedAt: (record.issuedAt as string) ?? null,
-          expiresAt: (record.expiresAt as string) ?? null,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed certification');
+    await indexRecord(db, COLLECTION, did, rkey, record);
   };
 }

src/jetstream/indexers/course.ts

···
 import type { Database } from '../../db/index.js';
-import { courses } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
+import { indexRecord, deleteRecord } from '../../services/record-indexer.js';
+
+const COLLECTION = 'id.sifa.profile.course';
 
 export function createCourseIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db.delete(courses).where(and(eq(courses.did, did), eq(courses.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted course');
+      await deleteRecord(db, COLLECTION, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(courses)
-      .values({
-        did,
-        rkey,
-        name: sanitize(record.name as string),
-        number: sanitizeOptional(record.number as string | undefined) ?? null,
-        institution: sanitizeOptional(record.institution as string | undefined) ?? null,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [courses.did, courses.rkey],
-        set: {
-          name: sanitize(record.name as string),
-          number: sanitizeOptional(record.number as string | undefined) ?? null,
-          institution: sanitizeOptional(record.institution as string | undefined) ?? null,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed course');
+    await indexRecord(db, COLLECTION, did, rkey, record);
   };
 }

src/jetstream/indexers/education.ts

···
 import type { Database } from '../../db/index.js';
-import { education } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
+import { indexEducation, deleteEducation } from '../../services/record-indexer.js';
 
 export function createEducationIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db.delete(education).where(and(eq(education.did, did), eq(education.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted education');
+      await deleteEducation(db, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(education)
-      .values({
-        did,
-        rkey,
-        institution: sanitize(record.institution as string),
-        institutionDid: (record.institutionDid as string) ?? null,
-        degree: sanitizeOptional(record.degree as string | undefined) ?? null,
-        fieldOfStudy: sanitizeOptional(record.fieldOfStudy as string | undefined) ?? null,
-        description: sanitizeOptional(record.description as string | undefined) ?? null,
-        startDate: (record.startDate as string) ?? null,
-        endDate: (record.endDate as string) ?? null,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [education.did, education.rkey],
-        set: {
-          institution: sanitize(record.institution as string),
-          institutionDid: (record.institutionDid as string) ?? null,
-          degree: sanitizeOptional(record.degree as string | undefined) ?? null,
-          fieldOfStudy: sanitizeOptional(record.fieldOfStudy as string | undefined) ?? null,
-          description: sanitizeOptional(record.description as string | undefined) ?? null,
-          startDate: (record.startDate as string) ?? null,
-          endDate: (record.endDate as string) ?? null,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed education');
+    await indexEducation(db, did, rkey, record);
   };
 }

src/jetstream/indexers/external-account.ts

···
 import type { Database } from '../../db/index.js';
-import { externalAccounts } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
+import { indexRecord, deleteRecord } from '../../services/record-indexer.js';
+
+const COLLECTION = 'id.sifa.profile.externalAccount';
 
 export function createExternalAccountIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db
-        .delete(externalAccounts)
-        .where(and(eq(externalAccounts.did, did), eq(externalAccounts.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted external account');
+      await deleteRecord(db, COLLECTION, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(externalAccounts)
-      .values({
-        did,
-        rkey,
-        platform: sanitize(record.platform as string),
-        url: sanitize(record.url as string),
-        label: sanitizeOptional(record.label as string | undefined) ?? null,
-        feedUrl: sanitizeOptional(record.feedUrl as string | undefined) ?? null,
-        isPrimary: (record.isPrimary as boolean) ?? false,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [externalAccounts.did, externalAccounts.rkey],
-        set: {
-          platform: sanitize(record.platform as string),
-          url: sanitize(record.url as string),
-          label: sanitizeOptional(record.label as string | undefined) ?? null,
-          feedUrl: sanitizeOptional(record.feedUrl as string | undefined) ?? null,
-          isPrimary: (record.isPrimary as boolean) ?? false,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed external account');
+    await indexRecord(db, COLLECTION, did, rkey, record);
   };
 }

src/jetstream/indexers/honor.ts

···
 import type { Database } from '../../db/index.js';
-import { honors } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
+import { indexRecord, deleteRecord } from '../../services/record-indexer.js';
+
+const COLLECTION = 'id.sifa.profile.honor';
 
 export function createHonorIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db.delete(honors).where(and(eq(honors.did, did), eq(honors.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted honor');
+      await deleteRecord(db, COLLECTION, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(honors)
-      .values({
-        did,
-        rkey,
-        title: sanitize(record.title as string),
-        issuer: sanitizeOptional(record.issuer as string | undefined) ?? null,
-        description: sanitizeOptional(record.description as string | undefined) ?? null,
-        awardedAt: (record.awardedAt as string) ?? null,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [honors.did, honors.rkey],
-        set: {
-          title: sanitize(record.title as string),
-          issuer: sanitizeOptional(record.issuer as string | undefined) ?? null,
-          description: sanitizeOptional(record.description as string | undefined) ?? null,
-          awardedAt: (record.awardedAt as string) ?? null,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed honor');
+    await indexRecord(db, COLLECTION, did, rkey, record);
   };
 }

src/jetstream/indexers/language.ts

···
 import type { Database } from '../../db/index.js';
-import { languages } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize } from '../../lib/sanitize.js';
+import { indexRecord, deleteRecord } from '../../services/record-indexer.js';
+
+const COLLECTION = 'id.sifa.profile.language';
 
 export function createLanguageIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db.delete(languages).where(and(eq(languages.did, did), eq(languages.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted language');
+      await deleteRecord(db, COLLECTION, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(languages)
-      .values({
-        did,
-        rkey,
-        name: sanitize(record.name as string),
-        proficiency: (record.proficiency as string) ?? null,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [languages.did, languages.rkey],
-        set: {
-          name: sanitize(record.name as string),
-          proficiency: (record.proficiency as string) ?? null,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed language');
+    await indexRecord(db, COLLECTION, did, rkey, record);
   };
 }

src/jetstream/indexers/position.ts

···
 import type { Database } from '../../db/index.js';
-import { positions, skillPositionLinks } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
-
-interface StrongRef {
-  uri: string;
-  cid: string;
-}
-
-/** Extract rkey from an AT Protocol URI: at://did/collection/rkey */
-function parseRkeyFromUri(uri: string): string | null {
-  const parts = uri.split('/');
-  return parts.length >= 5 ? (parts[4] ?? null) : null;
-}
-
-interface RecordLocation {
-  country?: string;
-  region?: string;
-  city?: string;
-  countryCode?: string;
-}
+import { indexPosition, deletePosition } from '../../services/record-indexer.js';
 
 export function createPositionIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db
-        .delete(skillPositionLinks)
-        .where(and(eq(skillPositionLinks.did, did), eq(skillPositionLinks.positionRkey, rkey)));
-      await db.delete(positions).where(and(eq(positions.did, did), eq(positions.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted position');
+      await deletePosition(db, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    const location = record.location as RecordLocation | undefined;
-
-    await db
-      .insert(positions)
-      .values({
-        did,
-        rkey,
-        companyName: sanitize(record.companyName as string),
-        companyDid: (record.companyDid as string) ?? null,
-        title: sanitize(record.title as string),
-        description: sanitizeOptional(record.description as string | undefined) ?? null,
-        employmentType: (record.employmentType as string) ?? null,
-        workplaceType: (record.workplaceType as string) ?? null,
-        locationCountry: sanitizeOptional(location?.country) ?? null,
-        locationRegion: sanitizeOptional(location?.region) ?? null,
-        locationCity: sanitizeOptional(location?.city) ?? null,
-        countryCode: sanitizeOptional(location?.countryCode) ?? null,
-        startDate: record.startDate as string,
-        endDate: (record.endDate as string) ?? null,
-        current: (record.current as boolean) ?? false,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [positions.did, positions.rkey],
-        set: {
-          companyName: sanitize(record.companyName as string),
-          companyDid: (record.companyDid as string) ?? null,
-          title: sanitize(record.title as string),
-          description: sanitizeOptional(record.description as string | undefined) ?? null,
-          employmentType: (record.employmentType as string) ?? null,
-          workplaceType: (record.workplaceType as string) ?? null,
-          locationCountry: sanitizeOptional(location?.country) ?? null,
-          locationRegion: sanitizeOptional(location?.region) ?? null,
-          locationCity: sanitizeOptional(location?.city) ?? null,
-          countryCode: sanitizeOptional(location?.countryCode) ?? null,
-          startDate: record.startDate as string,
-          endDate: (record.endDate as string) ?? null,
-          current: (record.current as boolean) ?? false,
-          indexedAt: new Date(),
-        },
-      });
-
-    // Sync skill-position links: delete-and-replace strategy
-    await db
-      .delete(skillPositionLinks)
-      .where(and(eq(skillPositionLinks.did, did), eq(skillPositionLinks.positionRkey, rkey)));
-
-    const skillRefs = record.skills as StrongRef[] | undefined;
-    if (skillRefs && Array.isArray(skillRefs) && skillRefs.length > 0) {
-      const linkValues = skillRefs
-        .map((ref) => {
-          const skillRkey = parseRkeyFromUri(ref.uri);
-          if (!skillRkey) {
-            logger.warn(
-              { did, rkey, uri: ref.uri },
-              'Could not parse skill rkey from strongRef URI',
-            );
-            return null;
-          }
-          return { did, positionRkey: rkey, skillRkey };
-        })
-        .filter((v): v is NonNullable<typeof v> => v !== null);
-
-      if (linkValues.length > 0) {
-        await db.insert(skillPositionLinks).values(linkValues).onConflictDoNothing();
-      }
-    }
-
-    logger.info({ did, rkey, operation, skillLinks: skillRefs?.length ?? 0 }, 'Indexed position');
+    await indexPosition(db, did, rkey, record);
   };
 }

src/jetstream/indexers/project.ts

···
 import type { Database } from '../../db/index.js';
-import { projects } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
+import { indexRecord, deleteRecord } from '../../services/record-indexer.js';
+
+const COLLECTION = 'id.sifa.profile.project';
 
 export function createProjectIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db.delete(projects).where(and(eq(projects.did, did), eq(projects.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted project');
+      await deleteRecord(db, COLLECTION, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(projects)
-      .values({
-        did,
-        rkey,
-        name: sanitize(record.name as string),
-        description: sanitizeOptional(record.description as string | undefined) ?? null,
-        url: (record.url as string) ?? null,
-        startedAt: (record.startedAt as string) ?? null,
-        endedAt: (record.endedAt as string) ?? null,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [projects.did, projects.rkey],
-        set: {
-          name: sanitize(record.name as string),
-          description: sanitizeOptional(record.description as string | undefined) ?? null,
-          url: (record.url as string) ?? null,
-          startedAt: (record.startedAt as string) ?? null,
-          endedAt: (record.endedAt as string) ?? null,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed project');
+    await indexRecord(db, COLLECTION, did, rkey, record);
   };
 }

src/jetstream/indexers/publication.ts

···
 import type { Database } from '../../db/index.js';
-import { publications } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
+import { indexRecord, deleteRecord } from '../../services/record-indexer.js';
+
+const COLLECTION = 'id.sifa.profile.publication';
 
 export function createPublicationIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db
-        .delete(publications)
-        .where(and(eq(publications.did, did), eq(publications.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted publication');
+      await deleteRecord(db, COLLECTION, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(publications)
-      .values({
-        did,
-        rkey,
-        title: sanitize(record.title as string),
-        publisher: sanitizeOptional(record.publisher as string | undefined) ?? null,
-        url: (record.url as string) ?? null,
-        description: sanitizeOptional(record.description as string | undefined) ?? null,
-        publishedAt: (record.publishedAt as string) ?? null,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [publications.did, publications.rkey],
-        set: {
-          title: sanitize(record.title as string),
-          publisher: sanitizeOptional(record.publisher as string | undefined) ?? null,
-          url: (record.url as string) ?? null,
-          description: sanitizeOptional(record.description as string | undefined) ?? null,
-          publishedAt: (record.publishedAt as string) ?? null,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed publication');
+    await indexRecord(db, COLLECTION, did, rkey, record);
   };
 }

src/jetstream/indexers/skill.ts

···
 import type { Database } from '../../db/index.js';
-import { skills, canonicalSkills } from '../../db/schema/index.js';
-import { and, eq, sql } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
-import { resolveSkill } from '../../services/skill-normalization.js';
+import { indexSkill, deleteSkill } from '../../services/record-indexer.js';
 
 export function createSkillIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      // Look up canonical_skill_id before deleting so we can decrement user_count
-      const existing = await db
-        .select({ canonicalSkillId: skills.canonicalSkillId })
-        .from(skills)
-        .where(and(eq(skills.did, did), eq(skills.rkey, rkey)))
-        .limit(1);
-
-      await db.delete(skills).where(and(eq(skills.did, did), eq(skills.rkey, rkey)));
-
-      if (existing[0]?.canonicalSkillId) {
-        await db
-          .update(canonicalSkills)
-          .set({ userCount: sql`GREATEST(${canonicalSkills.userCount} - 1, 0)` })
-          .where(eq(canonicalSkills.id, existing[0].canonicalSkillId));
-      }
-
-      logger.info({ did, rkey }, 'Deleted skill');
+      await deleteSkill(db, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    const skillName = sanitize(record.skillName as string);
-    const category = sanitizeOptional(record.category as string | undefined) ?? null;
-
-    // Run normalization pipeline
-    const canonical = await resolveSkill(db, skillName);
-    const canonicalSkillId = canonical?.id ?? null;
-
-    // Check if this is an update (existing record may already be linked to a different canonical)
-    const existing = await db
-      .select({ canonicalSkillId: skills.canonicalSkillId })
-      .from(skills)
-      .where(and(eq(skills.did, did), eq(skills.rkey, rkey)))
-      .limit(1);
-
-    const previousCanonicalId = existing[0]?.canonicalSkillId ?? null;
-
-    await db
-      .insert(skills)
-      .values({
-        did,
-        rkey,
-        skillName,
-        category,
-        canonicalSkillId,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [skills.did, skills.rkey],
-        set: {
-          skillName,
-          category,
-          canonicalSkillId,
-          indexedAt: new Date(),
-        },
-      });
-
-    // Update user_count: decrement old canonical (if changed), increment new
-    if (previousCanonicalId && previousCanonicalId !== canonicalSkillId) {
-      await db
-        .update(canonicalSkills)
-        .set({ userCount: sql`GREATEST(${canonicalSkills.userCount} - 1, 0)` })
-        .where(eq(canonicalSkills.id, previousCanonicalId));
-    }
-    if (canonicalSkillId && canonicalSkillId !== previousCanonicalId) {
-      await db
-        .update(canonicalSkills)
-        .set({ userCount: sql`${canonicalSkills.userCount} + 1` })
-        .where(eq(canonicalSkills.id, canonicalSkillId));
-    }
-
-    logger.info({ did, rkey, operation, canonicalSkillId }, 'Indexed skill');
+    await indexSkill(db, did, rkey, record);
   };
 }

src/jetstream/indexers/volunteering.ts

···
 import type { Database } from '../../db/index.js';
-import { volunteering } from '../../db/schema/index.js';
-import { and, eq } from 'drizzle-orm';
 import type { JetstreamEvent } from '../types.js';
-import { logger } from '../../logger.js';
-import { sanitize, sanitizeOptional } from '../../lib/sanitize.js';
+import { indexRecord, deleteRecord } from '../../services/record-indexer.js';
+
+const COLLECTION = 'id.sifa.profile.volunteering';
 
 export function createVolunteeringIndexer(db: Database) {
   return async (event: JetstreamEvent) => {
···
     const { operation, rkey, record } = commit;
 
     if (operation === 'delete') {
-      await db
-        .delete(volunteering)
-        .where(and(eq(volunteering.did, did), eq(volunteering.rkey, rkey)));
-      logger.info({ did, rkey }, 'Deleted volunteering');
+      await deleteRecord(db, COLLECTION, did, rkey);
       return;
     }
 
     if (!record) return;
-
-    await db
-      .insert(volunteering)
-      .values({
-        did,
-        rkey,
-        organization: sanitize(record.organization as string),
-        role: sanitizeOptional(record.role as string | undefined) ?? null,
-        cause: sanitizeOptional(record.cause as string | undefined) ?? null,
-        description: sanitizeOptional(record.description as string | undefined) ?? null,
-        startedAt: (record.startedAt as string) ?? null,
-        endedAt: (record.endedAt as string) ?? null,
-        createdAt: new Date(record.createdAt as string),
-        indexedAt: new Date(),
-      })
-      .onConflictDoUpdate({
-        target: [volunteering.did, volunteering.rkey],
-        set: {
-          organization: sanitize(record.organization as string),
-          role: sanitizeOptional(record.role as string | undefined) ?? null,
-          cause: sanitizeOptional(record.cause as string | undefined) ?? null,
-          description: sanitizeOptional(record.description as string | undefined) ?? null,
-          startedAt: (record.startedAt as string) ?? null,
-          endedAt: (record.endedAt as string) ?? null,
-          indexedAt: new Date(),
-        },
-      });
-
-    logger.info({ did, rkey, operation }, 'Indexed volunteering');
+    await indexRecord(db, COLLECTION, did, rkey, record);
   };
 }

src/routes/external-accounts.ts

···
 import { createAuthMiddleware, getAuthContext } from '../middleware/auth.js';
 import { discoverFeedUrl, fetchFeedItems } from '../services/feed-discovery.js';
 import { checkAndStoreVerification, isVerifiablePlatform } from '../services/verification.js';
+import { indexRecord, deleteRecord } from '../services/record-indexer.js';
 
 const FEED_CACHE_TTL = 1800; // 30 minutes
 
···
       await writeToUserPds(session, did, [
         buildApplyWritesOp('create', 'id.sifa.profile.externalAccount', rkey, record),
       ]);
+
+      await indexRecord(db, 'id.sifa.profile.externalAccount', did, rkey, record);
 
       // Trigger verification in the background
       const [profile] = await db.select().from(profiles).where(eq(profiles.did, did)).limit(1);
···
         buildApplyWritesOp('update', 'id.sifa.profile.externalAccount', rkey, record),
       ]);
 
+      await indexRecord(db, 'id.sifa.profile.externalAccount', did, rkey, record);
+
       // Re-verify in the background
       const [profile] = await db.select().from(profiles).where(eq(profiles.did, did)).limit(1);
       if (profile) {
···
         ]);
       } catch (err) {
         if (isPdsRecordNotFound(err)) {
+          await deleteRecord(db, 'id.sifa.profile.externalAccount', did, rkey);
           return reply.status(200).send({ ok: true });
         }
         return handlePdsError(err, reply);
       }
+
+      await deleteRecord(db, 'id.sifa.profile.externalAccount', did, rkey);
 
       return reply.status(200).send({ ok: true });
     },

src/routes/profile-write.ts

···
 import { sanitize, sanitizeOptional } from '../lib/sanitize.js';
 import { wipeSifaData } from '../services/profile-wipe.js';
 import {
+  indexSkill,
+  deleteSkill,
+  indexPosition,
+  deletePosition,
+  indexEducation,
+  deleteEducation,
+  indexRecord,
+  deleteRecord,
+} from '../services/record-indexer.js';
+import {
   sessions as sessionsTable,
   oauthSessions as oauthSessionsTable,
 } from '../db/schema/index.js';
···
       buildApplyWritesOp('create', 'id.sifa.profile.position', rkey, record),
     ]);
 
+    // Write-through: index in local DB immediately
+    await indexPosition(db, did, rkey, record);
+
     return reply.status(201).send({ rkey });
   });
 
···
         buildApplyWritesOp('update', 'id.sifa.profile.position', rkey, record),
       ]);
 
+      // Write-through: update local DB immediately
+      await indexPosition(db, did, rkey, record);
+
       return reply.status(200).send({ ok: true });
     },
   );
···
         ]);
       } catch (err) {
         if (isPdsRecordNotFound(err)) {
+          await deletePosition(db, did, rkey);
           return reply.status(200).send({ ok: true });
         }
         return handlePdsError(err, reply);
       }
 
+      // Write-through: delete from local DB immediately
+      await deletePosition(db, did, rkey);
+
       return reply.status(200).send({ ok: true });
     },
   );
···
     await writeToUserPds(session, did, [
       buildApplyWritesOp('create', 'id.sifa.profile.education', rkey, record),
     ]);
+
+    await indexEducation(db, did, rkey, record);
 
     return reply.status(201).send({ rkey });
   });
···
       await writeToUserPds(session, did, [
         buildApplyWritesOp('update', 'id.sifa.profile.education', rkey, record),
       ]);
+
+      await indexEducation(db, did, rkey, record);
 
       return reply.status(200).send({ ok: true });
     },
···
         ]);
       } catch (err) {
         if (isPdsRecordNotFound(err)) {
+          await deleteEducation(db, did, rkey);
           return reply.status(200).send({ ok: true });
         }
         return handlePdsError(err, reply);
       }
+
+      await deleteEducation(db, did, rkey);
 
       return reply.status(200).send({ ok: true });
     },
···
     await writeToUserPds(session, did, [
       buildApplyWritesOp('create', 'id.sifa.profile.skill', rkey, record),
     ]);
+
+    await indexSkill(db, did, rkey, record);
 
     return reply.status(201).send({ rkey });
   });
···
         buildApplyWritesOp('update', 'id.sifa.profile.skill', rkey, record),
       ]);
 
+      await indexSkill(db, did, rkey, record);
+
       return reply.status(200).send({ ok: true });
     },
   );
···
         ]);
       } catch (err) {
         if (isPdsRecordNotFound(err)) {
+          await deleteSkill(db, did, rkey);
           return reply.status(200).send({ ok: true });
         }
         return handlePdsError(err, reply);
       }
+
+      await deleteSkill(db, did, rkey);
 
       return reply.status(200).send({ ok: true });
     },
···
       const data = parsed.data as Record<string, unknown>;
       const record: Record<string, unknown> = { createdAt: new Date().toISOString(), ...data };
       await writeToUserPds(session, did, [buildApplyWritesOp('create', collection, rkey, record)]);
+      await indexRecord(db, collection, did, rkey, record);
       return reply.status(201).send({ rkey });
     },
   );
···
       const data = parsed.data as Record<string, unknown>;
       const record: Record<string, unknown> = { createdAt: new Date().toISOString(), ...data };
       await writeToUserPds(session, did, [buildApplyWritesOp('update', collection, rkey, record)]);
+      await indexRecord(db, collection, did, rkey, record);
       return reply.status(200).send({ ok: true });
     },
   );
···
         await writeToUserPds(session, did, [buildApplyWritesOp('delete', collection, rkey)]);
       } catch (err) {
         if (isPdsRecordNotFound(err)) {
+          await deleteRecord(db, collection, did, rkey);
           return reply.status(200).send({ ok: true });
         }
         return handlePdsError(err, reply);
       }
+      await deleteRecord(db, collection, did, rkey);
       return reply.status(200).send({ ok: true });
     },
   );

src/services/record-indexer.ts

···
+import type { Database } from '../db/index.js';
+import {
+  skills,
+  positions,
+  education,
+  certifications,
+  projects,
+  volunteering,
+  publications,
+  courses,
+  honors,
+  languages,
+  externalAccounts,
+  canonicalSkills,
+  skillPositionLinks,
+} from '../db/schema/index.js';
+import { and, eq, sql } from 'drizzle-orm';
+import { sanitize, sanitizeOptional } from '../lib/sanitize.js';
+import { resolveSkill } from './skill-normalization.js';
+import { logger } from '../logger.js';
+
+// --- Skill ---
+
+export async function indexSkill(
+  db: Database,
+  did: string,
+  rkey: string,
+  record: Record<string, unknown>,
+): Promise<void> {
+  const skillName = sanitize(record.skillName as string);
+  const category = sanitizeOptional(record.category as string | undefined) ?? null;
+
+  const canonical = await resolveSkill(db, skillName);
+  const canonicalSkillId = canonical?.id ?? null;
+
+  const existing = await db
+    .select({ canonicalSkillId: skills.canonicalSkillId })
+    .from(skills)
+    .where(and(eq(skills.did, did), eq(skills.rkey, rkey)))
+    .limit(1);
+
+  const previousCanonicalId = existing[0]?.canonicalSkillId ?? null;
+
+  await db
+    .insert(skills)
+    .values({
+      did,
+      rkey,
+      skillName,
+      category,
+      canonicalSkillId,
+      createdAt: new Date(record.createdAt as string),
+      indexedAt: new Date(),
+    })
+    .onConflictDoUpdate({
+      target: [skills.did, skills.rkey],
+      set: {
+        skillName,
+        category,
+        canonicalSkillId,
+        indexedAt: new Date(),
+      },
+    });
+
+  if (previousCanonicalId && previousCanonicalId !== canonicalSkillId) {
+    await db
+      .update(canonicalSkills)
+      .set({ userCount: sql`GREATEST(${canonicalSkills.userCount} - 1, 0)` })
+      .where(eq(canonicalSkills.id, previousCanonicalId));
+  }
+  if (canonicalSkillId && canonicalSkillId !== previousCanonicalId) {
+    await db
+      .update(canonicalSkills)
+      .set({ userCount: sql`${canonicalSkills.userCount} + 1` })
+      .where(eq(canonicalSkills.id, canonicalSkillId));
+  }
+
+  logger.info({ did, rkey, canonicalSkillId }, 'Indexed skill (write-through)');
+}
+
+export async function deleteSkill(db: Database, did: string, rkey: string): Promise<void> {
+  const existing = await db
+    .select({ canonicalSkillId: skills.canonicalSkillId })
+    .from(skills)
+    .where(and(eq(skills.did, did), eq(skills.rkey, rkey)))
+    .limit(1);
+
+  await db.delete(skills).where(and(eq(skills.did, did), eq(skills.rkey, rkey)));
+
+  if (existing[0]?.canonicalSkillId) {
+    await db
+      .update(canonicalSkills)
+      .set({ userCount: sql`GREATEST(${canonicalSkills.userCount} - 1, 0)` })
+      .where(eq(canonicalSkills.id, existing[0].canonicalSkillId));
+  }
+
+  logger.info({ did, rkey }, 'Deleted skill (write-through)');
+}
+
+// --- Position ---
+
+function parseRkeyFromUri(uri: string): string | null {
+  const parts = uri.split('/');
+  return parts.length >= 5 ? (parts[4] ?? null) : null;
+}
+
+interface StrongRef {
+  uri: string;
+  cid: string;
+}
+
+interface RecordLocation {
+  country?: string;
+  region?: string;
+  city?: string;
+  countryCode?: string;
+}
+
+export async function indexPosition(
+  db: Database,
+  did: string,
+  rkey: string,
+  record: Record<string, unknown>,
+): Promise<void> {
+  const location = record.location as RecordLocation | undefined;
+
+  await db
+    .insert(positions)
+    .values({
+      did,
+      rkey,
+      companyName: sanitize(record.companyName as string),
+      companyDid: (record.companyDid as string) ?? null,
+      title: sanitize(record.title as string),
+      description: sanitizeOptional(record.description as string | undefined) ?? null,
+      employmentType: (record.employmentType as string) ?? null,
+      workplaceType: (record.workplaceType as string) ?? null,
+      locationCountry: sanitizeOptional(location?.country) ?? null,
+      locationRegion: sanitizeOptional(location?.region) ?? null,
+      locationCity: sanitizeOptional(location?.city) ?? null,
+      countryCode: sanitizeOptional(location?.countryCode) ?? null,
+      startDate: record.startDate as string,
+      endDate: (record.endDate as string) ?? null,
+      current: (record.current as boolean) ?? false,
+      createdAt: new Date(record.createdAt as string),
+      indexedAt: new Date(),
+    })
+    .onConflictDoUpdate({
+      target: [positions.did, positions.rkey],
+      set: {
+        companyName: sanitize(record.companyName as string),
+        companyDid: (record.companyDid as string) ?? null,
+        title: sanitize(record.title as string),
+        description: sanitizeOptional(record.description as string | undefined) ?? null,
+        employmentType: (record.employmentType as string) ?? null,
+        workplaceType: (record.workplaceType as string) ?? null,
+        locationCountry: sanitizeOptional(location?.country) ?? null,
+        locationRegion: sanitizeOptional(location?.region) ?? null,
+        locationCity: sanitizeOptional(location?.city) ?? null,
+        countryCode: sanitizeOptional(location?.countryCode) ?? null,
+        startDate: record.startDate as string,
+        endDate: (record.endDate as string) ?? null,
+        current: (record.current as boolean) ?? false,
+        indexedAt: new Date(),
+      },
+    });
+
+  // Sync skill-position links: delete-and-replace
+  await db
+    .delete(skillPositionLinks)
+    .where(and(eq(skillPositionLinks.did, did), eq(skillPositionLinks.positionRkey, rkey)));
+
+  const skillRefs = record.skills as StrongRef[] | undefined;
+  if (skillRefs && Array.isArray(skillRefs) && skillRefs.length > 0) {
+    const linkValues = skillRefs
+      .map((ref) => {
+        const skillRkey = parseRkeyFromUri(ref.uri);
+        if (!skillRkey) {
+          logger.warn({ did, rkey, uri: ref.uri }, 'Could not parse skill rkey from strongRef URI');
+          return null;
+        }
+        return { did, positionRkey: rkey, skillRkey };
+      })
+      .filter((v): v is NonNullable<typeof v> => v !== null);
+
+    if (linkValues.length > 0) {
+      await db.insert(skillPositionLinks).values(linkValues).onConflictDoNothing();
+    }
+  }
+
+  logger.info(
+    { did, rkey, skillLinks: skillRefs?.length ?? 0 },
+    'Indexed position (write-through)',
+  );
+}
+
+export async function deletePosition(db: Database, did: string, rkey: string): Promise<void> {
+  await db
+    .delete(skillPositionLinks)
+    .where(and(eq(skillPositionLinks.did, did), eq(skillPositionLinks.positionRkey, rkey)));
+  await db.delete(positions).where(and(eq(positions.did, did), eq(positions.rkey, rkey)));
+  logger.info({ did, rkey }, 'Deleted position (write-through)');
+}
+
+// --- Education ---
+
+export async function indexEducation(
+  db: Database,
+  did: string,
+  rkey: string,
+  record: Record<string, unknown>,
+): Promise<void> {
+  await db
+    .insert(education)
+    .values({
+      did,
+      rkey,
+      institution: sanitize(record.institution as string),
+      institutionDid: (record.institutionDid as string) ?? null,
+      degree: sanitizeOptional(record.degree as string | undefined) ?? null,
+      fieldOfStudy: sanitizeOptional(record.fieldOfStudy as string | undefined) ?? null,
+      description: sanitizeOptional(record.description as string | undefined) ?? null,
+      startDate: (record.startDate as string) ?? null,
+      endDate: (record.endDate as string) ?? null,
+      createdAt: new Date(record.createdAt as string),
+      indexedAt: new Date(),
+    })
+    .onConflictDoUpdate({
+      target: [education.did, education.rkey],
+      set: {
+        institution: sanitize(record.institution as string),
+        institutionDid: (record.institutionDid as string) ?? null,
+        degree: sanitizeOptional(record.degree as string | undefined) ?? null,
+        fieldOfStudy: sanitizeOptional(record.fieldOfStudy as string | undefined) ?? null,
+        description: sanitizeOptional(record.description as string | undefined) ?? null,
+        startDate: (record.startDate as string) ?? null,
+        endDate: (record.endDate as string) ?? null,
+        indexedAt: new Date(),
+      },
+    });
+
+  logger.info({ did, rkey }, 'Indexed education (write-through)');
+}
+
+export async function deleteEducation(db: Database, did: string, rkey: string): Promise<void> {
+  await db.delete(education).where(and(eq(education.did, did), eq(education.rkey, rkey)));
+  logger.info({ did, rkey }, 'Deleted education (write-through)');
+}
+
+// --- Generic record indexer for remaining collections ---
+
+const COLLECTION_INDEXERS: Record<
+  string,
+  {
+    table:
+      | typeof certifications
+      | typeof projects
+      | typeof volunteering
+      | typeof publications
+      | typeof courses
+      | typeof honors
+      | typeof languages
+      | typeof externalAccounts;
+    fields: (record: Record<string, unknown>) => Record<string, unknown>;
+  }
+> = {
+  'id.sifa.profile.certification': {
+    table: certifications,
+    fields: (r) => ({
+      name: sanitize(r.name as string),
+      authority: sanitizeOptional(r.authority as string | undefined) ?? null,
+      credentialId: (r.credentialId as string) ?? null,
+      credentialUrl: (r.credentialUrl as string) ?? null,
+      issuedAt: (r.issuedAt as string) ?? null,
+      expiresAt: (r.expiresAt as string) ?? null,
+    }),
+  },
+  'id.sifa.profile.project': {
+    table: projects,
+    fields: (r) => ({
+      name: sanitize(r.name as string),
+      description: sanitizeOptional(r.description as string | undefined) ?? null,
+      url: (r.url as string) ?? null,
+      startedAt: (r.startedAt as string) ?? null,
+      endedAt: (r.endedAt as string) ?? null,
+    }),
+  },
+  'id.sifa.profile.volunteering': {
+    table: volunteering,
+    fields: (r) => ({
+      organization: sanitize(r.organization as string),
+      role: sanitizeOptional(r.role as string | undefined) ?? null,
+      cause: sanitizeOptional(r.cause as string | undefined) ?? null,
+      description: sanitizeOptional(r.description as string | undefined) ?? null,
+      startedAt: (r.startedAt as string) ?? null,
+      endedAt: (r.endedAt as string) ?? null,
+    }),
+  },
+  'id.sifa.profile.publication': {
+    table: publications,
+    fields: (r) => ({
+      title: sanitize(r.title as string),
+      publisher: sanitizeOptional(r.publisher as string | undefined) ?? null,
+      url: (r.url as string) ?? null,
+      description: sanitizeOptional(r.description as string | undefined) ?? null,
+      publishedAt: (r.publishedAt as string) ?? null,
+    }),
+  },
+  'id.sifa.profile.course': {
+    table: courses,
+    fields: (r) => ({
+      name: sanitize(r.name as string),
+      number: sanitizeOptional(r.number as string | undefined) ?? null,
+      institution: sanitizeOptional(r.institution as string | undefined) ?? null,
+    }),
+  },
+  'id.sifa.profile.honor': {
+    table: honors,
+    fields: (r) => ({
+      title: sanitize(r.title as string),
+      issuer: sanitizeOptional(r.issuer as string | undefined) ?? null,
+      description: sanitizeOptional(r.description as string | undefined) ?? null,
+      awardedAt: (r.awardedAt as string) ?? null,
+    }),
+  },
+  'id.sifa.profile.language': {
+    table: languages,
+    fields: (r) => ({
+      name: sanitize(r.name as string),
+      proficiency: (r.proficiency as string) ?? null,
+    }),
+  },
+  'id.sifa.profile.externalAccount': {
+    table: externalAccounts,
+    fields: (r) => ({
+      platform: sanitize(r.platform as string),
+      url: sanitize(r.url as string),
+      label: sanitizeOptional(r.label as string | undefined) ?? null,
+      feedUrl: sanitizeOptional(r.feedUrl as string | undefined) ?? null,
+      isPrimary: (r.isPrimary as boolean) ?? false,
+    }),
+  },
+};
+
+export async function indexRecord(
+  db: Database,
+  collection: string,
+  did: string,
+  rkey: string,
+  record: Record<string, unknown>,
+): Promise<void> {
+  const config = COLLECTION_INDEXERS[collection];
+  if (!config) {
+    logger.warn({ collection }, 'No indexer config for collection');
+    return;
+  }
+
+  const fields = config.fields(record);
+  const table = config.table as typeof certifications; // all have same (did, rkey) conflict target shape
+
+  await db
+    .insert(table)
+    .values({
+      did,
+      rkey,
+      ...fields,
+      createdAt: new Date(record.createdAt as string),
+      indexedAt: new Date(),
+    } as typeof table.$inferInsert)
+    .onConflictDoUpdate({
+      target: [table.did, table.rkey],
+      set: {
+        ...fields,
+        indexedAt: new Date(),
+      } as Partial<typeof table.$inferInsert>,
+    });
+
+  logger.info({ did, rkey, collection }, 'Indexed record (write-through)');
+}
+
+export async function deleteRecord(
+  db: Database,
+  collection: string,
+  did: string,
+  rkey: string,
+): Promise<void> {
+  const config = COLLECTION_INDEXERS[collection];
+  if (!config) {
+    logger.warn({ collection }, 'No indexer config for collection');
+    return;
+  }
+
+  const table = config.table as typeof certifications;
+  await db.delete(table).where(and(eq(table.did, did), eq(table.rkey, rkey)));
+  logger.info({ did, rkey, collection }, 'Deleted record (write-through)');
+}

tests/services/record-indexer.test.ts

···
+import { describe, it, expect, beforeAll, afterAll, afterEach } from 'vitest';
+import { createDb } from '../../src/db/index.js';
+import {
+  skills,
+  positions,
+  education,
+  certifications,
+  projects,
+  volunteering,
+  publications,
+  courses,
+  honors,
+  languages,
+  canonicalSkills,
+  profiles,
+  skillPositionLinks,
+} from '../../src/db/schema/index.js';
+import { eq, and, sql } from 'drizzle-orm';
+
+// These functions don't exist yet -- this test will fail (RED)
+import {
+  indexSkill,
+  deleteSkill,
+  indexPosition,
+  deletePosition,
+  indexEducation,
+  deleteEducation,
+  indexRecord,
+  deleteRecord,
+} from '../../src/services/record-indexer.js';
+
+describe('Record indexer service', () => {
+  const db = createDb(process.env.DATABASE_URL ?? 'postgresql://sifa:sifa@localhost:5432/sifa');
+  const testDid = 'did:plc:write-through-test';
+
+  beforeAll(async () => {
+    await db
+      .insert(profiles)
+      .values({
+        did: testDid,
+        handle: 'write-through-test.bsky.social',
+        createdAt: new Date(),
+      })
+      .onConflictDoNothing();
+
+    await db
+      .insert(canonicalSkills)
+      .values({
+        canonicalName: 'WriteThrough TestLang',
+        slug: 'writethrough-testlang',
+        category: 'technical',
+        aliases: ['wt-test', 'writethrough-testlang'],
+        userCount: 0,
+      })
+      .onConflictDoNothing();
+  });
+
+  afterAll(async () => {
+    await db.delete(skillPositionLinks).where(eq(skillPositionLinks.did, testDid));
+    await db.delete(skills).where(eq(skills.did, testDid));
+    await db.delete(positions).where(eq(positions.did, testDid));
+    await db.delete(education).where(eq(education.did, testDid));
+    await db.delete(certifications).where(eq(certifications.did, testDid));
+    await db.delete(projects).where(eq(projects.did, testDid));
+    await db.delete(volunteering).where(eq(volunteering.did, testDid));
+    await db.delete(publications).where(eq(publications.did, testDid));
+    await db.delete(courses).where(eq(courses.did, testDid));
+    await db.delete(honors).where(eq(honors.did, testDid));
+    await db.delete(languages).where(eq(languages.did, testDid));
+    await db.execute(
+      sql`DELETE FROM canonical_skills WHERE slug = 'writethrough-testlang' AND canonical_name = 'WriteThrough TestLang'`,
+    );
+    await db.execute(sql`DELETE FROM profiles WHERE did = ${testDid}`);
+    await db.$client.end();
+  });
+
+  // --- Skill indexing ---
+
+  describe('indexSkill', () => {
+    afterEach(async () => {
+      await db.delete(skills).where(eq(skills.did, testDid));
+    });
+
+    it('creates a skill record in local DB', async () => {
+      await indexSkill(db, testDid, '3wt-skill-1', {
+        skillName: 'TypeScript',
+        category: 'technical',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(skills)
+        .where(and(eq(skills.did, testDid), eq(skills.rkey, '3wt-skill-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].skillName).toBe('TypeScript');
+      expect(rows[0].category).toBe('technical');
+    });
+
+    it('resolves canonical skill when alias matches', async () => {
+      await indexSkill(db, testDid, '3wt-skill-2', {
+        skillName: 'WT-Test',
+        category: 'technical',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(skills)
+        .where(and(eq(skills.did, testDid), eq(skills.rkey, '3wt-skill-2')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].canonicalSkillId).not.toBeNull();
+    });
+
+    it('is idempotent -- second call updates, does not duplicate', async () => {
+      await indexSkill(db, testDid, '3wt-skill-3', {
+        skillName: 'React',
+        category: 'technical',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+      await indexSkill(db, testDid, '3wt-skill-3', {
+        skillName: 'React',
+        category: 'technical',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(skills)
+        .where(and(eq(skills.did, testDid), eq(skills.rkey, '3wt-skill-3')));
+      expect(rows).toHaveLength(1);
+    });
+  });
+
+  describe('deleteSkill', () => {
+    it('removes a skill record from local DB', async () => {
+      await indexSkill(db, testDid, '3wt-skill-del', {
+        skillName: 'Python',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      await deleteSkill(db, testDid, '3wt-skill-del');
+
+      const rows = await db
+        .select()
+        .from(skills)
+        .where(and(eq(skills.did, testDid), eq(skills.rkey, '3wt-skill-del')));
+      expect(rows).toHaveLength(0);
+    });
+
+    it('is idempotent -- deleting non-existent record does not throw', async () => {
+      await expect(deleteSkill(db, testDid, '3wt-nonexistent')).resolves.not.toThrow();
+    });
+  });
+
+  // --- Position indexing ---
+
+  describe('indexPosition', () => {
+    afterEach(async () => {
+      await db.delete(skillPositionLinks).where(eq(skillPositionLinks.did, testDid));
+      await db.delete(positions).where(eq(positions.did, testDid));
+    });
+
+    it('creates a position record in local DB', async () => {
+      await indexPosition(db, testDid, '3wt-pos-1', {
+        companyName: 'Acme Corp',
+        title: 'Senior Engineer',
+        startDate: '2024-01',
+        current: true,
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(positions)
+        .where(and(eq(positions.did, testDid), eq(positions.rkey, '3wt-pos-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].companyName).toBe('Acme Corp');
+      expect(rows[0].title).toBe('Senior Engineer');
+      expect(rows[0].current).toBe(true);
+    });
+
+    it('syncs skill-position links', async () => {
+      // First create the skill
+      await indexSkill(db, testDid, '3wt-link-skill', {
+        skillName: 'Go',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      await indexPosition(db, testDid, '3wt-pos-link', {
+        companyName: 'GoLand',
+        title: 'Go Dev',
+        startDate: '2024-01',
+        current: false,
+        skills: [{ uri: `at://${testDid}/id.sifa.profile.skill/3wt-link-skill`, cid: 'bafyfake' }],
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const links = await db
+        .select()
+        .from(skillPositionLinks)
+        .where(
+          and(
+            eq(skillPositionLinks.did, testDid),
+            eq(skillPositionLinks.positionRkey, '3wt-pos-link'),
+          ),
+        );
+      expect(links).toHaveLength(1);
+      expect(links[0].skillRkey).toBe('3wt-link-skill');
+    });
+
+    it('is idempotent', async () => {
+      const record = {
+        companyName: 'Dupe Corp',
+        title: 'Eng',
+        startDate: '2024-01',
+        current: false,
+        createdAt: '2026-03-16T00:00:00Z',
+      };
+      await indexPosition(db, testDid, '3wt-pos-idem', record);
+      await indexPosition(db, testDid, '3wt-pos-idem', record);
+
+      const rows = await db
+        .select()
+        .from(positions)
+        .where(and(eq(positions.did, testDid), eq(positions.rkey, '3wt-pos-idem')));
+      expect(rows).toHaveLength(1);
+    });
+  });
+
+  describe('deletePosition', () => {
+    it('removes position and its skill links', async () => {
+      await indexPosition(db, testDid, '3wt-pos-del', {
+        companyName: 'Del Corp',
+        title: 'Eng',
+        startDate: '2024-01',
+        current: false,
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      await deletePosition(db, testDid, '3wt-pos-del');
+
+      const rows = await db
+        .select()
+        .from(positions)
+        .where(and(eq(positions.did, testDid), eq(positions.rkey, '3wt-pos-del')));
+      expect(rows).toHaveLength(0);
+    });
+  });
+
+  // --- Education indexing ---
+
+  describe('indexEducation', () => {
+    afterEach(async () => {
+      await db.delete(education).where(eq(education.did, testDid));
+    });
+
+    it('creates an education record in local DB', async () => {
+      await indexEducation(db, testDid, '3wt-edu-1', {
+        institution: 'MIT',
+        degree: 'MSc',
+        fieldOfStudy: 'Computer Science',
+        startDate: '2020-09',
+        endDate: '2022-06',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(education)
+        .where(and(eq(education.did, testDid), eq(education.rkey, '3wt-edu-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].institution).toBe('MIT');
+      expect(rows[0].degree).toBe('MSc');
+    });
+  });
+
+  describe('deleteEducation', () => {
+    it('removes an education record', async () => {
+      await indexEducation(db, testDid, '3wt-edu-del', {
+        institution: 'Oxford',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+      await deleteEducation(db, testDid, '3wt-edu-del');
+
+      const rows = await db
+        .select()
+        .from(education)
+        .where(and(eq(education.did, testDid), eq(education.rkey, '3wt-edu-del')));
+      expect(rows).toHaveLength(0);
+    });
+  });
+
+  // --- Generic indexRecord for remaining collections ---
+
+  describe('indexRecord (generic)', () => {
+    it('creates a certification record', async () => {
+      await indexRecord(db, 'id.sifa.profile.certification', testDid, '3wt-cert-1', {
+        name: 'AWS Solutions Architect',
+        authority: 'Amazon',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(certifications)
+        .where(and(eq(certifications.did, testDid), eq(certifications.rkey, '3wt-cert-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].name).toBe('AWS Solutions Architect');
+    });
+
+    it('creates a project record', async () => {
+      await indexRecord(db, 'id.sifa.profile.project', testDid, '3wt-proj-1', {
+        name: 'Open Source Tool',
+        url: 'https://example.com',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(projects)
+        .where(and(eq(projects.did, testDid), eq(projects.rkey, '3wt-proj-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].name).toBe('Open Source Tool');
+    });
+
+    it('creates a course record', async () => {
+      await indexRecord(db, 'id.sifa.profile.course', testDid, '3wt-course-1', {
+        name: 'Distributed Systems',
+        institution: 'MIT OCW',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(courses)
+        .where(and(eq(courses.did, testDid), eq(courses.rkey, '3wt-course-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].name).toBe('Distributed Systems');
+    });
+
+    it('creates a volunteering record', async () => {
+      await indexRecord(db, 'id.sifa.profile.volunteering', testDid, '3wt-vol-1', {
+        organization: 'Code for All',
+        role: 'Mentor',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(volunteering)
+        .where(and(eq(volunteering.did, testDid), eq(volunteering.rkey, '3wt-vol-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].organization).toBe('Code for All');
+    });
+
+    it('creates a publication record', async () => {
+      await indexRecord(db, 'id.sifa.profile.publication', testDid, '3wt-pub-1', {
+        title: 'My Paper',
+        publisher: 'IEEE',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(publications)
+        .where(and(eq(publications.did, testDid), eq(publications.rkey, '3wt-pub-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].title).toBe('My Paper');
+    });
+
+    it('creates an honor record', async () => {
+      await indexRecord(db, 'id.sifa.profile.honor', testDid, '3wt-honor-1', {
+        title: 'Best Paper Award',
+        issuer: 'ACM',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(honors)
+        .where(and(eq(honors.did, testDid), eq(honors.rkey, '3wt-honor-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].title).toBe('Best Paper Award');
+    });
+
+    it('creates a language record', async () => {
+      await indexRecord(db, 'id.sifa.profile.language', testDid, '3wt-lang-1', {
+        name: 'Dutch',
+        proficiency: 'native',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+
+      const rows = await db
+        .select()
+        .from(languages)
+        .where(and(eq(languages.did, testDid), eq(languages.rkey, '3wt-lang-1')));
+      expect(rows).toHaveLength(1);
+      expect(rows[0].name).toBe('Dutch');
+    });
+
+    it('is idempotent for generic records', async () => {
+      const record = { name: 'Idempotent Cert', createdAt: '2026-03-16T00:00:00Z' };
+      await indexRecord(db, 'id.sifa.profile.certification', testDid, '3wt-cert-idem', record);
+      await indexRecord(db, 'id.sifa.profile.certification', testDid, '3wt-cert-idem', record);
+
+      const rows = await db
+        .select()
+        .from(certifications)
+        .where(and(eq(certifications.did, testDid), eq(certifications.rkey, '3wt-cert-idem')));
+      expect(rows).toHaveLength(1);
+    });
+  });
+
+  describe('deleteRecord (generic)', () => {
+    it('deletes a certification record', async () => {
+      await indexRecord(db, 'id.sifa.profile.certification', testDid, '3wt-cert-del', {
+        name: 'To Delete',
+        createdAt: '2026-03-16T00:00:00Z',
+      });
+      await deleteRecord(db, 'id.sifa.profile.certification', testDid, '3wt-cert-del');
+
+      const rows = await db
+        .select()
+        .from(certifications)
+        .where(and(eq(certifications.did, testDid), eq(certifications.rkey, '3wt-cert-del')));
+      expect(rows).toHaveLength(0);
+    });
+
+    it('is idempotent -- deleting non-existent record does not throw', async () => {
+      await expect(
+        deleteRecord(db, 'id.sifa.profile.certification', testDid, '3wt-nonexistent'),
+      ).resolves.not.toThrow();
+    });
+  });
+});