MailGPT

export class GraphUpdater {
  private stores: EntityStores;
  private reconciler: DriftReconciler;

  async processEmail(event: EmailMetadataEvent): Promise<GraphUpdateResult> {
    const idempotencyKey = `process:${event.messageId}`;
    if (await this.stores.hasBeenApplied(idempotencyKey)) {
      return { status: "duplicate", idempotencyKey };
    }

    const thread = await this.resolveThread(event.threadId, event);
    const sender = await this.resolvePerson(event.from, event);
    const recipients = await Promise.all(event.to.map((e) => this.resolvePerson(e, event)));

    const email: Email = {
      schema_version: 2,
      message_id: event.messageId,
      thread_id: event.threadId,
      subject: event.subject,
      sent_at: event.timestamp,
      received_at: Date.now(),
      from: event.from, to: event.to, cc: event.cc,
      in_reply_to: event.inReplyTo ?? null,
      references: event.references ?? [],
      size_bytes: event.sizeBytes,
      checksum: "",
      created_at: Date.now(),
    };
    email.checksum = computeEntityChecksum(email);

    const edges: EdgeMutation[] = [
      { from: sender.email, to: email.message_id, type: "SENT", properties: { timestamp: event.timestamp } },
      ...recipients.map((r) => ({ from: r.email, to: email.message_id, type: "RECEIVED", properties: { timestamp: event.timestamp } })),
      { from: email.message_id, to: event.threadId, type: "BELONGS_TO", properties: {} },
    ];

    await this.stores.transaction(async (tx) => {
      await tx.storeEntity("email", email);
      await tx.storeEntity("thread", thread);
      await tx.storeEntity("person", sender);
      for (const r of recipients) await tx.storeEntity("person", r);
      for (const edge of edges) await tx.storeEdge(edge);
      await tx.recordIdempotency(idempotencyKey);
    });

    this.reconciler.scheduleCheck("email", email.message_id).catch(console.error);
    return { status: "applied", idempotencyKey, edgesCreated: edges.length };
  }
}