Function createEntityStore

src/core/entity-store.ts:97–205

(opts: EntityStoreOptions = {})

}

export function createEntityStore(opts: EntityStoreOptions = {}): IEntityStore {
  const filePath = opts.filePath ?? ENTITY_FILE
  const emitter = new EventEmitter()
  emitter.setMaxListeners(50)

  // Serialize mutating ops so each read-modify-write-rename cycle is atomic for
  // this store instance. Without it, concurrent upserts — which Pi triggers by
  // running tool calls in PARALLEL — race on the shared `${filePath}.tmp`,
  // interleaving bytes and corrupting the file (observed in the wild as
  // "Unexpected token ','" on the next read), and clobber each other's writes.
  let writeTail: Promise<unknown> = Promise.resolve()
  function serialize<T>(op: () => Promise<T>): Promise<T> {
    const result = writeTail.then(op, op)
    writeTail = result.then(() => undefined, () => undefined)
    return result
  }

  async function readAll(): Promise<Entity[]> {
    let raw: string
    try {
      raw = await readFile(filePath, 'utf-8')
    } catch (err: unknown) {
      if (err instanceof Error && 'code' in err && (err as NodeJS.ErrnoException).code === 'ENOENT') {
        return []
      }
      throw err
    }
    const out: Entity[] = []
    for (const line of raw.split('\n')) {
      if (!line.trim()) continue
      try {
        out.push(JSON.parse(line) as Entity)
      } catch {
        // Tolerate a malformed line instead of bricking every entity op. A
        // corrupted line (e.g. left by a pre-fix concurrent-write interleave)
        // is skipped here and dropped on the next atomic rewrite — self-healing.
        console.warn('entity-store: skipping malformed line in', filePath)
      }
    }
    return out
  }

  async function writeAll(entities: Entity[]): Promise<void> {
    await mkdir(dirname(filePath), { recursive: true })
    const body = entities.length > 0 ? entities.map((e) => JSON.stringify(e)).join('\n') + '\n' : ''
    // Atomic rewrite — tmp + rename, same as InboxStore.delete. Current-state
    // file (deduped by name), not an append log, so upsert is a read-modify-write.
    const tmp = `${filePath}.tmp`
    await writeFile(tmp, body, 'utf-8')
    await rename(tmp, filePath)
  }

  async function upsert(input: EntityInput): Promise<Entity> {
    validateInput(input) // sync — fail fast, outside the write queue
    return serialize(async () => {
      const all = await readAll()
      const k = keyOf(input.name)
      const idx = all.findIndex((e) => keyOf(e.name) === k)