IaC / SwiftFlow

Artifact be9a71937c21f8685fdf6c7b7db46fd48970548067dd9120d48a987dcc2cd06a:

• File backend/src/sync_products.gleam — part of check-in [6c065e93fd] at 2026-10-23 04:32:35 on branch trunk — 库存只采集配置中的仓库，默认总仓。

  未列入的仓库仍保留名称用于显示，不再同步其出入库单据。 (user: pluto size: 5701)

import app.{type Context}
import clock
import database
import gleam/list
import gleam/order
import gleam/result
import gleam/string
import ql361api.{type GoodsItem}
import ql361api/goods
import repo.{SyncRun}
import session
import sync_lock
import warehouse_scope
import sync_report.{type SyncError, type SyncReport, SyncBusy, SyncDb, SyncNotConfigured, SyncQl361, SyncRejected, SyncReport}
import wisp

pub type Mode {
  Incremental
  Full
}

const max_pages = 500

pub fn run(
  ctx: Context,
  trigger: String,
  mode: Mode,
) -> Result(SyncReport, SyncError) {
  case configured(ctx) {
    False -> Error(SyncNotConfigured)
    True ->
      case sync_lock.with_lock(ctx.lock, "products", fn() { do_sync(ctx, trigger, mode) }) {
        Error(Nil) -> Error(SyncBusy)
        Ok(result) -> result
      }
  }
}

fn do_sync(ctx: Context, trigger: String, mode: Mode) -> Result(SyncReport, SyncError) {
  let started = clock.now_local()
  use previous <- result_db(database.transaction(ctx.db, fn(conn) {
    repo.get_cursor(conn, "products")
  }))
  let modified_from = case mode {
    Full -> ""
    Incremental -> previous
  }
  let modified_to = clock.now_local()
  use items <- bind_sync(
    ctx,
    trigger,
    started,
    collect_pages(ctx, modified_from, modified_to, 0, []),
  )
  use warehouses <- bind_sync(
    ctx,
    trigger,
    started,
    case session.with_client(ctx.session, ql361api.fetch_warehouses) {
      Ok(rows) -> Ok(rows)
      Error(err) -> Error(SyncQl361(err))
    },
  )

  let products = list.filter(items, fn(item) { !item.is_category })
  let cursor = case mode {
    Full -> max_modified("", items)
    Incremental -> max_modified(previous, items)
  }
  let cursor = case cursor {
    "" -> previous
    value -> value
  }
  let finished = clock.now_local()
  let run =
    SyncRun(
      id: run_id(),
      entity: "products",
      trigger: trigger,
      started_at: started,
      finished_at: finished,
      status: "ok",
      fetched: list.length(items),
      upserted: list.length(products),
      error_message: "",
    )

  let written =
    database.transaction(ctx.db, fn(conn) {
      let stock_ids =
        warehouse_scope.stock_warehouse_ids(warehouses, ctx.config.stock_warehouses)
      use _ <- result.try(list.try_fold(warehouses, Nil, fn(_, warehouse) {
        repo.upsert_warehouse(
          conn,
          warehouse,
          list.contains(stock_ids, warehouse.id),
          finished,
        )
      }))
      use _ <- result.try(list.try_fold(products, Nil, fn(_, item) {
        repo.upsert_product(conn, item, finished)
      }))
      use _ <- result.try(repo.set_cursor(conn, "products", cursor, finished))
      repo.insert_run(conn, run)
    })

  case written {
    Error(err) -> {
      record_failure(ctx, trigger, started, database_message(err))
      Error(SyncDb(err))
    }
    Ok(_) ->
      Ok(SyncReport(
        entity: "products",
        trigger: trigger,
        fetched: list.length(items),
        upserted: list.length(products),
        cursor: cursor,
      ))
  }
}

fn collect_pages(
  ctx: Context,
  modified_from: String,
  modified_to: String,
  page: Int,
  acc: List(GoodsItem),
) -> Result(List(GoodsItem), SyncError) {
  case page >= max_pages {
    True -> Error(SyncRejected("商品分页超过 500 页，已中止且未推进游标"))
    False -> {
      let fetched =
        session.with_client(ctx.session, fn(client) {
          ql361api.fetch_goods_page(
            client,
            page,
            goods.page_size,
            modified_from,
            modified_to,
          )
        })
      case fetched {
        Error(err) -> Error(SyncQl361(err))
        Ok(page_data) -> {
          let acc = list.append(acc, page_data.items)
          let count = list.length(page_data.items)
          let finished =
            count == 0
            || count < goods.page_size
            || list.length(acc) >= page_data.total
          case finished {
            True -> Ok(acc)
            False ->
              collect_pages(ctx, modified_from, modified_to, page + 1, acc)
          }
        }
      }
    }
  }
}

pub fn max_modified(current: String, items: List(GoodsItem)) -> String {
  list.fold(items, current, fn(acc, item) {
    case item.modified_at != "" && string.compare(item.modified_at, acc) == order.Gt {
      True -> item.modified_at
      False -> acc
    }
  })
}

fn record_failure(ctx: Context, trigger: String, started: String, message: String) -> Nil {
  let run =
    SyncRun(
      id: run_id(),
      entity: "products",
      trigger: trigger,
      started_at: started,
      finished_at: clock.now_local(),
      status: "error",
      fetched: 0,
      upserted: 0,
      error_message: message,
    )
  let _ = database.transaction(ctx.db, fn(conn) { repo.insert_run(conn, run) })
  Nil
}

fn configured(ctx: Context) -> Bool {
  ctx.config.ql361_username != "" && ctx.config.ql361_password != ""
}

fn run_id() -> String {
  clock.now_local() <> "-" <> wisp.random_string(8)
}

fn database_message(error: database.DbError) -> String {
  case error {
    database.DbError(message) -> message
  }
}

fn result_db(
  result: Result(a, database.DbError),
  next: fn(a) -> Result(b, SyncError),
) -> Result(b, SyncError) {
  case result {
    Ok(value) -> next(value)
    Error(err) -> Error(SyncDb(err))
  }
}

fn bind_sync(
  ctx: Context,
  trigger: String,
  started: String,
  result: Result(a, SyncError),
  next: fn(a) -> Result(b, SyncError),
) -> Result(b, SyncError) {
  case result {
    Ok(value) -> next(value)
    Error(err) -> {
      record_failure(ctx, trigger, started, sync_report.describe(err))
      Error(err)
    }
  }
}