オフラインで書いた投稿が、復帰した瞬間にもう一つ増える——再送に強い送信アウトボックスの設計

// outbox/store.ts
import * as SQLite from 'expo-sqlite';
 
const db = SQLite.openDatabaseSync('outbox.db');
 
db.execSync(`
  CREATE TABLE IF NOT EXISTS outbox (
    localId TEXT PRIMARY KEY,
    idempotencyKey TEXT NOT NULL,
    endpoint TEXT NOT NULL,
    payload TEXT NOT NULL,
    status TEXT NOT NULL,
    attempts INTEGER NOT NULL,
    nextAttemptAt INTEGER NOT NULL,
    dependsOn TEXT,
    createdAt INTEGER NOT NULL
  );
`);
 
export function insertItem(item: OutboxItem): void {
  db.runSync(
    `INSERT INTO outbox VALUES (?,?,?,?,?,?,?,?,?)`,
    item.localId, item.idempotencyKey, item.endpoint,
    JSON.stringify(item.payload), item.status, item.attempts,
    item.nextAttemptAt, item.dependsOn ?? null, item.createdAt,
  );
}
 
export function dueItems(now: number): OutboxItem[] {
  const rows = db.getAllSync<any>(
    `SELECT * FROM outbox
     WHERE status IN ('pending','failed') AND nextAttemptAt <= ?
     ORDER BY createdAt ASC`, now,
  );
  return rows.map((r) => ({ ...r, payload: JSON.parse(r.payload) }));
}

// outbox/remap.ts
// 親の送信成功時に呼ぶ。tempId -> serverId をキュー全体へ反映する。
export function remapId(tempId: string, serverId: string): void {
  const children = db.getAllSync<any>(
    `SELECT * FROM outbox WHERE payload LIKE ?`, `%${tempId}%`,
  );
  for (const row of children) {
    const next = JSON.parse(row.payload as string);
    // payload を再帰的に走査して tempId を serverId へ置換
    const replaced = JSON.parse(
      JSON.stringify(next).split(tempId).join(serverId),
    );
    db.runSync(
      `UPDATE outbox SET payload = ?, dependsOn = NULL WHERE localId = ?`,
      JSON.stringify(replaced), row.localId,
    );
  }
}

// outbox/runner.ts
const MAX_ATTEMPTS = 6;
 
function isRetryable(status: number): boolean {
  return status === 0 || status === 429 || status >= 500;
}
 
async function flushOne(item: OutboxItem): Promise<void> {
  db.runSync(`UPDATE outbox SET status='inflight' WHERE localId=?`, item.localId);
  try {
    const res = await send(item);
    if (res.ok) {
      const body = await res.json();
      if (body.id && item.payload && (item.payload as any).localId) {
        remapId((item.payload as any).localId, body.id);
      }
      db.runSync(`DELETE FROM outbox WHERE localId=?`, item.localId);
      return;
    }
    if (!isRetryable(res.status)) {
      // 何度送っても直らない → 隔離して通知
      db.runSync(`UPDATE outbox SET status='dead' WHERE localId=?`, item.localId);
      notifyUserOfDeadItem(item);
      return;
    }
    throw new Error(`retryable ${res.status}`);
  } catch (e) {
    const attempts = item.attempts + 1;
    const status = attempts >= MAX_ATTEMPTS ? 'dead' : 'failed';
    const backoff = Math.min(2 ** attempts * 1000, 5 * 60 * 1000); // 上限5分
    db.runSync(
      `UPDATE outbox SET status=?, attempts=?, nextAttemptAt=? WHERE localId=?`,
      status, attempts, Date.now() + backoff, item.localId,
    );
  }
}

// outbox/loop.ts
let running = false;
 
export async function flush(): Promise<void> {
  if (running) return;        // 単一実行ロック
  running = true;
  try {
    let batch = dueItems(Date.now()).filter((i) => !i.dependsOn);
    for (const item of batch) {
      await flushOne(item);    // 直列に流す
    }
  } finally {
    running = false;
  }
}