use hashed rkeys to prevent duplicate contributions

Author: trezy

src/contributions/client.ts

@@ -104,10 +104,33 @@ export class ContributionClient {
 		}
 	}
 
+	async getRecord(collection: string, rkey: string): Promise<Record<string, unknown> | null> {
+		if (!this.session) {
+			throw new Error('Not authenticated — call restore() first')
+		}
+
+		const response = await fetch(
+			`${this.instanceUrl}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(this.session.did)}&collection=${encodeURIComponent(collection)}&rkey=${encodeURIComponent(rkey)}`,
+			{ headers: { 'X-Client-Key': this.clientKey } },
+		)
+
+		if (response.status === 404 || response.status === 400) {
+			return null
+		}
+
+		if (!response.ok) {
+			const body = await response.text()
+			throw new Error(`getRecord returned ${response.status}: ${body}`)
+		}
+
+		return await response.json() as Record<string, unknown>
+	}
+
 	async createContribution(params: {
 		contributionType: 'correction' | 'addition' | 'newGame'
 		changes: Record<string, unknown>
 		subject?: string
+		rkey?: string
 	}): Promise<ContributionResult> {
 		if (!this.session) {
 			throw new Error('Not authenticated — call restore() first')
@@ -120,6 +143,9 @@ export class ContributionClient {
 		if (params.subject) {
 			body.subject = params.subject
 		}
+		if (params.rkey) {
+			body.rkey = params.rkey
+		}
 
 		const response = await this.session.fetchHandler(
 			`${this.instanceUrl}/xrpc/games.gamesgamesgamesgames.createContribution`,

src/contributions/handler.ts

@@ -1,5 +1,8 @@
 import type { ContributionClient } from './client.js'
 import { diffGameRecord } from './diff.js'
+import { contentHashRkey } from './hash.js'
+
+const CONTRIBUTION_COLLECTION = 'games.gamesgamesgamesgames.contribution'
 
 export interface ContributionSyncResult {
 	action: 'created' | 'skipped'
@@ -22,9 +25,16 @@ export async function syncGameViaContribution(
 	const existing = await contributionClient.getGameByIgdbId(igdbId)
 
 	if (!existing) {
+		const rkey = contentHashRkey(mappedRecord)
+		const existingRecord = await contributionClient.getRecord(CONTRIBUTION_COLLECTION, rkey)
+		if (existingRecord) {
+			return { action: 'skipped' }
+		}
+
 		const result = await contributionClient.createContribution({
 			contributionType: 'newGame',
 			changes: mappedRecord,
+			rkey,
 		})
 		return { action: 'created', contributionType: 'newGame', uri: result.uri }
 	}
@@ -35,11 +45,18 @@ export async function syncGameViaContribution(
 		return { action: 'skipped' }
 	}
 
+	const rkey = contentHashRkey(changes)
+	const existingRecord = await contributionClient.getRecord(CONTRIBUTION_COLLECTION, rkey)
+	if (existingRecord) {
+		return { action: 'skipped' }
+	}
+
 	const subjectUri = existing.uri
 	const result = await contributionClient.createContribution({
 		contributionType: 'correction',
 		changes,
 		subject: subjectUri,
+		rkey,
 	})
 
 	if (existing.redirectedFrom) {

src/contributions/hash.ts

@@ -0,0 +1,21 @@
+import { createHash } from 'node:crypto'
+
+function stableStringify(value: unknown): string {
+	if (value === null || value === undefined) return 'null'
+	if (typeof value !== 'object') return JSON.stringify(value)
+
+	if (Array.isArray(value)) {
+		return '[' + value.map(stableStringify).join(',') + ']'
+	}
+
+	const sorted = Object.keys(value as Record<string, unknown>).sort()
+	const entries = sorted.map(
+		(k) => JSON.stringify(k) + ':' + stableStringify((value as Record<string, unknown>)[k]),
+	)
+	return '{' + entries.join(',') + '}'
+}
+
+export function contentHashRkey(changes: Record<string, unknown>): string {
+	const serialized = stableStringify(changes)
+	return createHash('sha256').update(serialized).digest('hex').slice(0, 32)
+}

src/sync.ts

@@ -13,7 +13,6 @@
 
 import 'dotenv/config'
 
-import Database from 'better-sqlite3'
 import { IGDBClient } from './igdb/client.js'
 import { AtprotoClient } from './atproto/client.js'
 import { StateManager } from './state.js'
@@ -97,28 +96,6 @@ async function main() {
 	console.log(`Started at ${new Date().toISOString()}`)
 	console.log()
 
-	// Validate state vs PDS to prevent duplicate creation
-	const pdsStorePath = process.env.PDS_STORE_PATH ?? '/pds/actors/29/did:web:gamesgamesgamesgames.games/store.sqlite'
-	try {
-		const pdsDb = new Database(pdsStorePath, { readonly: true })
-		const pdsRow = pdsDb.prepare(
-			"SELECT COUNT(*) as count FROM record WHERE collection = 'games.gamesgamesgamesgames.game'",
-		).get() as { count: number }
-		pdsDb.close()
-
-		const stateCount = state.getEntityCount('game')
-		console.log(`[sync] PDS game records: ${pdsRow.count}, State entities: ${stateCount}`)
-
-		if (pdsRow.count > 0 && stateCount < pdsRow.count * 0.9) {
-			console.error(`[sync] ABORT: state.sqlite is out of sync with PDS (state=${stateCount}, PDS=${pdsRow.count}).`)
-			console.error('[sync] This would cause duplicate record creation. Run cleanup first.')
-			process.exit(1)
-		}
-	} catch (err) {
-		console.warn(`[sync] Could not validate PDS store at ${pdsStorePath}: ${(err as Error).message}`)
-		console.warn('[sync] Continuing without validation (set PDS_STORE_PATH if running on server)')
-	}
-
 	// Sync in dependency order
 
 	// // 1. Platform Families