ap-bridge / lib / state.js

// state.js — durable publisher state: which post rkeys have already been delivered.
//
// This MUST survive a restart: lose it and the bridge either re-delivers everything (spamming
// followers) or re-primes to "everything already sent" and silently drops whatever was posted
// while it was down. Write-then-rename so a crash cannot truncate it.
//
// A MAP of rkey -> projection fingerprint, not a set and not a high-watermark (see lib/publisher.js
// for why ordering-based tracking was wrong). The fingerprint is what lets us tell "already sent
// and unchanged" from "already sent but the projection has since improved" — the latter becomes an
// Update{Note}, which is exactly how Mastodon models an edit.
//
// Bounded to the most recent N: an rkey that falls off the end can never reappear, because the
// drain only ever looks at the current tip of the repo.

import fs from 'bare-fs'
import { dirname } from 'bare-path'

const MAX = 500

export class JsonState {
  constructor (path) { this._path = path; this._data = null }

  _load () {
    if (this._data) return this._data
    try { this._data = JSON.parse(fs.readFileSync(this._path, 'utf8')) } catch { this._data = {} }
    // Migration: the first version stored a bare array of rkeys. Convert to a map with an EMPTY
    // fingerprint, which reads as "delivered, projection unknown" — the drain then sends one
    // Update per post to bring receivers up to the current projection, once, and records the real
    // fingerprint. Self-healing and bounded by what listNotes returns.
    if (Array.isArray(this._data.delivered)) {
      this._data.delivered = Object.fromEntries(this._data.delivered.map((k) => [k, '']))
    }
    if (!this._data.delivered || typeof this._data.delivered !== 'object') this._data.delivered = {}
    return this._data
  }

  _flush () {
    fs.mkdirSync(dirname(this._path), { recursive: true })
    const tmp = this._path + '.tmp'
    fs.writeFileSync(tmp, JSON.stringify(this._data, null, 2))
    fs.renameSync(tmp, this._path)
  }

  async getProfileFingerprint () { return this._load().profileFp || '' }
  async setProfileFingerprint (v) { this._load().profileFp = v; this._flush() }
  async isPrimed () { return !!this._load().primed }
  async setPrimed () { this._load().primed = true; this._flush() }
  async getDelivered () { return new Map(Object.entries(this._load().delivered)) }

  async markDelivered (entries) {
    const d = this._load()
    for (const [k, fp] of entries) if (k) d.delivered[k] = fp || ''
    const keys = Object.keys(d.delivered)
    if (keys.length > MAX) {
      for (const k of keys.slice(0, keys.length - MAX)) delete d.delivered[k]
    }
    this._flush()
  }
}

export class MemoryState {
  constructor (delivered = [], primed = false) {
    this._set = new Map(delivered.map((d) => Array.isArray(d) ? d : [d, '']))
    this._primed = primed
  }
  async getProfileFingerprint () { return this._fp || '' }
  async setProfileFingerprint (v) { this._fp = v }
  async isPrimed () { return this._primed }
  async setPrimed () { this._primed = true }
  async getDelivered () { return new Map(this._set) }
  async markDelivered (entries) { for (const [k, fp] of entries) if (k) this._set.set(k, fp || '') }
}