IaC / SwiftFlow

Artifact 50e3bef4bcbd4f0c7f3203aee195e9ee5eea8ac23f09d7dd85a1cf70fcc29ea8:

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

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

import app.{type Context}
import clock
import config
import database
import gleam/list
import gleam/option.{Some}
import gleam/result
import ql361api.{type SalesOrder, type SalesOrderSummary}
import ql361api/orders as ql_orders
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

const max_pages = 500

pub fn run(ctx: Context, trigger: String) -> Result(SyncReport, SyncError) {
  case ctx.config.ql361_username != "" && ctx.config.ql361_password != "" {
    False -> Error(SyncNotConfigured)
    True ->
      case sync_lock.with_lock(ctx.lock, "orders", fn() { do_sync(ctx, trigger) }) {
        Error(Nil) -> Error(SyncBusy)
        Ok(result) -> result
      }
  }
}

fn do_sync(ctx: Context, trigger: String) -> Result(SyncReport, SyncError) {
  let started = clock.now_local()
  let #(begin_date, end_date) = config.order_window(ctx.config)
  let orders_result = collect_pages(ctx, begin_date, end_date, 0, [])
  let warehouses_result = case orders_result {
    Error(_) -> Error(SyncRejected(""))
    Ok(_) ->
      case session.with_client(ctx.session, ql361api.fetch_warehouses) {
        Ok(rows) -> Ok(rows)
        Error(err) -> Error(SyncQl361(err))
      }
  }

  case orders_result, warehouses_result {
    Error(err), _ -> {
      record_failure(ctx, trigger, started, sync_report.describe(err))
      Error(err)
    }
    _, Error(err) -> {
      record_failure(ctx, trigger, started, sync_report.describe(err))
      Error(err)
    }
    Ok(orders), Ok(warehouses) ->
      write_orders(ctx, trigger, started, begin_date, end_date, orders, warehouses)
  }
}

fn write_orders(
  ctx: Context,
  trigger: String,
  started: String,
  begin_date: String,
  end_date: String,
  orders: List(SalesOrderSummary),
  warehouses: List(ql361api.QlWarehouse),
) -> Result(SyncReport, SyncError) {
  let tokens = ctx.config.stock_warehouses
  let stock_ids = warehouse_scope.stock_warehouse_ids(warehouses, tokens)
  let orders =
    list.filter(orders, fn(order) {
      warehouse_scope.bill_in_stock_scope(
        order.warehouse_id,
        order.warehouse_name,
        warehouses,
        tokens,
      )
    })
  let prepared =
    database.transaction(ctx.db, fn(conn) {
      use pending <- result.try(repo.list_pending_bill_ids(
        conn,
        begin_date,
        end_date,
        stock_ids,
      ))
      use pairs <- result.try(list.try_map(orders, fn(order) {
        use stamp <- result.try(repo.get_order_stamp(conn, order.bill_id))
        Ok(#(order, stamp))
      }))
      Ok(#(pending, pairs))
    })

  use #(pending, pairs) <- result_db(prepared)
  let need =
    list.filter(pairs, fn(pair) { repo.needs_detail(pair.1, pair.0) })
    |> list.map(fn(pair) { pair.0.bill_id })
  let seen = list.map(orders, fn(order) { order.bill_id })
  let missing =
    list.filter(pending, fn(bill_id) { !list.contains(seen, bill_id) })
  let #(details, detail_errors) =
    fetch_details(ctx, list.append(need, missing), [], [])

  let finished = clock.now_local()
  let status = case detail_errors {
    [] -> "ok"
    _ -> "error"
  }
  let error_message = case detail_errors {
    [] -> ""
    [err, ..] -> sync_report.describe(err)
  }
  let run =
    SyncRun(
      id: clock.now_local() <> "-" <> wisp.random_string(8),
      entity: "orders",
      trigger: trigger,
      started_at: started,
      finished_at: finished,
      status: status,
      fetched: list.length(orders),
      upserted: list.length(orders),
      error_message: error_message,
    )

  let written =
    database.transaction(ctx.db, fn(conn) {
      use _ <- result.try(repo.delete_documents_outside_warehouses(conn, stock_ids))
      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(orders, Nil, fn(_, order) {
        repo.upsert_order_summary(conn, order, finished)
      }))
      use _ <- result.try(list.try_fold(details, Nil, fn(_, order) {
        repo.save_order_detail(conn, order, finished)
      }))
      use _ <- result.try(repo.set_cursor(conn, "orders", finished, finished))
      repo.insert_run(conn, run)
    })

  case written, detail_errors {
    Error(err), _ -> Error(SyncDb(err))
    Ok(_), [] ->
      Ok(SyncReport(
        entity: "orders",
        trigger: trigger,
        fetched: list.length(orders),
        upserted: list.length(orders),
        cursor: finished,
      ))
    Ok(_), [err, ..] -> Error(err)
  }
}

fn collect_pages(
  ctx: Context,
  begin_date: String,
  end_date: String,
  page: Int,
  acc: List(SalesOrderSummary),
) -> Result(List(SalesOrderSummary), SyncError) {
  case page >= max_pages {
    True -> Error(SyncRejected("订单分页超过 500 页，已中止"))
    False -> {
      let fetched =
        session.with_client(ctx.session, fn(client) {
          ql361api.fetch_sales_order_page(
            client,
            begin_date,
            end_date,
            Some(20),
            page,
            ql_orders.page_size,
          )
        })
      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 < ql_orders.page_size
            || list.length(acc) >= page_data.total
          case finished {
            True -> Ok(acc)
            False -> collect_pages(ctx, begin_date, end_date, page + 1, acc)
          }
        }
      }
    }
  }
}

fn fetch_details(
  ctx: Context,
  bill_ids: List(Int),
  ok: List(SalesOrder),
  errors: List(SyncError),
) -> #(List(SalesOrder), List(SyncError)) {
  case bill_ids {
    [] -> #(list.reverse(ok), list.reverse(errors))
    [bill_id, ..rest] -> {
      let fetched =
        session.with_client(ctx.session, fn(client) {
          ql361api.fetch_order_detail(client, bill_id)
        })
      case fetched {
        Ok(order) -> fetch_details(ctx, rest, [order, ..ok], errors)
        Error(err) -> fetch_details(ctx, rest, ok, [SyncQl361(err), ..errors])
      }
    }
  }
}

fn record_failure(
  ctx: Context,
  trigger: String,
  started: String,
  message: String,
) -> Nil {
  let run =
    SyncRun(
      id: clock.now_local() <> "-" <> wisp.random_string(8),
      entity: "orders",
      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 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))
  }
}