PocketBaseで軽量なワークフロー状態管理を実現する — Temporal/Airflowを使わない選択肢

class PocketBase {
  private api: AxiosInstance;

  constructor() {
    this.api = axios.create({
      baseURL: "http://127.0.0.1:8090",
    });

    // axiosインターセプターで自動接続チェック + 認証
    this.api.interceptors.request.use(async (config) => {
      await this.checkConnection();
      config.headers.Authorization = `Bearer ${await this.getAuthToken()}`;
      return config;
    });
  }

  private async checkConnection(): Promise<void> {
    const isRunning = await this.isServerRunning();
    if (isRunning) return;
    await this.initPocketBase();
  }

  public initPocketBase(): Promise<void> {
    return new Promise<void>((resolve, reject) => {
      if (environment.CURRENT_OS === "darwin") {
        execSync(`chmod +x "${pocketbasePath}"`);
        execSync(`xattr -d com.apple.quarantine "${pocketbasePath}"`);
      }

      const process = spawn(pocketbasePath, ["serve"]);

      const timeout = setTimeout(() => {
        process.kill();
        reject(new Error("PocketBase startup timed out"));
      }, 10000);

      const onData = (chunk: Buffer) => {
        if (chunk.toString().includes("Server started at")) {
          clearTimeout(timeout);
          process.stdout.removeListener("data", onData);
          process.stderr.removeListener("data", onData);
          resolve();
        }
      };

      process.stdout.on("data", onData);
      process.stderr.on("data", onData);

      process.on("error", (err) => {
        clearTimeout(timeout);
        reject(err);
      });
    });
  }

  private async getAuthToken(): Promise<string> {
    // インターセプター内でのループを避けるため、axiosを直接使用
    const response = await axios.post(
      `${this.pocketbaseUrl}/api/admins/auth-with-password`,
      {
        identity: credentials.POCKETBASE_USERNAME,
        password: credentials.POCKETBASE_PASSWORD,
      }
    );
    return response.data.token;
  }

  // ステータスでフィルタリング（カスタムエンドポイント）
  public async filterByStatus(
    status: string,
    equipmentType: string,
    daysAfter: number = 0
  ): Promise<EquipmentExchange[]> {
    const response = await this.api.post(
      "/api/collections/EquipmentExchange/filterByStatus",
      { status, equipmentType, daysAfter }
    );
    return response.data.items;
  }

  // ステータス更新
  public async updateStatus(serialNumber: string, status: string): Promise<void> {
    await this.api.post("/api/collections/EquipmentExchange/updateStatus", {
      serialNumber,
      status,
    });
  }

  // 重複チェック付きの一括登録
  public async registerEntries(entries: EquipmentExchange[]): Promise<void> {
    for (const entry of entries) {
      if (await this.checkForDuplicate(entry.serialNumber)) {
        Logger.error(`Duplicate found for ${entry.serialNumber}, skipping.`);
        continue;
      }
      await this.api.post("/api/collections/EquipmentExchange/records", entry);
    }
  }

// ステータス + 機器種別 + 日付条件で複合フィルタリング
routerAdd("POST", "api/collections/EquipmentExchange/filterByStatus", (c) => {
  const status = $apis.requestInfo(c).data.status;
  const equipmentType = $apis.requestInfo(c).data.equipmentType;
  const daysAfter = $apis.requestInfo(c).data.daysAfter;
  const systemTime = $apis.requestInfo(c).data.systemTime;

  // DynamicModelで結果のスキーマを定義
  const records = arrayOf(
    new DynamicModel({
      serialNumber: "",
      accountId: "",
      contactName: "",
      shipmentAddress: "",
      deliveryDate: "",
      processStatus: "",
      equipmentType: "",
      trackingNumber: "",
      requestDate: "",
      lastCheckDate: "",
      retry: "",
    })
  );

  // 生SQLで複雑なフィルタリング
  const query = [
    `SELECT *`,
    `FROM EquipmentExchange`,
    `WHERE processStatus = '${status}'`,
    `AND equipmentType = '${equipmentType}'`,
    `AND retry < 3`,
    daysAfter === 0 ? "" : `AND '${systemTime}' >= deliveryDate`,
    daysAfter === 0
      ? ""
      : `AND '${systemTime}' >= datetime(requestDate, '+${daysAfter} days')`,
    `ORDER BY created ASC`,
  ].join(" ");

  $app.dao().db().newQuery(query).all(records);

  return c.json(200, { items: records });
});

// ステータス更新
routerAdd("POST", "api/collections/EquipmentExchange/updateStatus", (c) => {
  const serialNumber = $apis.requestInfo(c).data.serialNumber;
  const status = $apis.requestInfo(c).data.status;
  const query = [
    `UPDATE EquipmentExchange`,
    `SET processStatus = '${status}'`,
    `WHERE serialNumber = '${serialNumber}'`,
  ].join(" ");

  $app.dao().db().newQuery(query).execute();

  return c.json(200, { message: "query executed successfully." });
});

// リトライカウンターのインクリメント
routerAdd("POST", "api/collections/EquipmentExchange/incRetry", (c) => {
  const serialNumber = $apis.requestInfo(c).data.serialNumber;
  const query = [
    `UPDATE EquipmentExchange`,
    `SET retry = retry + 1`,
    `WHERE serialNumber = '${serialNumber}'`,
  ].join(" ");

  $app.dao().db().newQuery(query).execute();

  return c.json(200, { message: "query executed successfully." });
});

// レコード削除
routerAdd("POST", "api/collections/EquipmentExchange/deleteRecord", (c) => {
  const serialNumber = $apis.requestInfo(c).data.serialNumber;
  const query = `DELETE FROM EquipmentExchange WHERE serialNumber = '${serialNumber}'`;

  $app.dao().db().newQuery(query).execute();

  return c.json(200, { message: "query executed successfully." });
});

// 同一配送先・同一日付のエントリをグループ化してバッチ処理用にマーク
routerAdd("POST", "api/collections/EquipmentExchange/groupEntries", (c) => {
  const records = arrayOf(
    new DynamicModel({
      shipmentAddress: "",
      deliveryDate: "",
      entryCount: "",
    })
  );

  const query = [
    `WITH GroupedEntries AS (`,
    `  SELECT shipmentAddress, deliveryDate, COUNT(*) as entryCount`,
    `  FROM EquipmentExchange`,
    `  WHERE processStatus = 'INIT'`,
    `  GROUP BY shipmentAddress, deliveryDate`,
    `  HAVING COUNT(*) > 1`,
    `)`,
    `UPDATE EquipmentExchange`,
    `SET processStatus = 'BATCH_READY'`,
    `FROM GroupedEntries`,
    `WHERE EquipmentExchange.shipmentAddress = GroupedEntries.shipmentAddress`,
    `  AND EquipmentExchange.deliveryDate = GroupedEntries.deliveryDate`,
    `  AND EquipmentExchange.processStatus = 'INIT';`,
  ].join(" ");

  $app.dao().db().newQuery(query).all(records);

  return c.json(200, { items: records });
});

  public async writeLog(log: ExchangeLog): Promise<void> {
    await this.api.post("/api/collections/ExchangeLogs/records", log);
  }

  public async deleteApiLogs(entries: { id: string }[]): Promise<void> {
    const deletePromises = entries.map((record) =>
      this.api.delete(`/api/collections/ExchangeLogs/records/${record.id}`)
    );
    await Promise.all(deletePromises); // 並列削除
  }