IaC / SwiftFlow

Artifact e1f9262e3f41b771c20052a16234bb71da5613adcec06e91d58eedc9b589110a:

• File backend/src/repo.gleam — part of check-in [d49666a1ca] at 2026-10-23 06:33:00 on branch trunk — 波次组单，按到期日占用库存并排出拣货路线

  - 制单员把同一作业仓的待发货订单合成波次，相同商品汇总后优先占用更早到期的批次 - 拣货行按库位编码排列，缺货单独标出；组单和撤销都不扣库存 - 未完成波次会占住推荐数量，避免下一张波次重复分配 (user: pluto size: 64086)

import batch_code
import database.{
  type Conn, type DbError, DbError, FloatParam, IntParam, TextParam,
}
import gleam/dynamic/decode
import gleam/float
import gleam/list
import gleam/option.{type Option, None, Some}
import gleam/result
import gleam/string
import location.{type Location, Location}
import ql361api/models.{
  type GoodsItem, type PurchaseOrder, type PurchaseOrderItem,
  type PurchaseOrderSummary, type QlWarehouse, type SalesOrder, type SalesOrderItem,
  type SalesOrderSummary, GoodsItem, PurchaseOrder, PurchaseOrderItem,
  PurchaseOrderSummary, QlWarehouse, SalesOrder, SalesOrderItem, SalesOrderSummary,
}
import ql361api/json_codec
import rbac

pub type InboundPage {
  InboundPage(items: List(PurchaseOrderSummary), total: Int)
}

pub type ReceiptLine {
  ReceiptLine(
    line_no: Int,
    bill_detail_id: Int,
    sku_id: Int,
    goods_code: String,
    goods_name: String,
    barcode: String,
    received_qty: Float,
    batch_no: String,
    produced_on: String,
    expires_on: String,
    location_code: String,
  )
}

pub type Receipt {
  Receipt(
    id: String,
    bill_id: Int,
    warehouse_id: Int,
    created_at: String,
    lines: List(ReceiptLine),
  )
}

pub type InventoryBatch {
  InventoryBatch(
    warehouse_id: Int,
    location_code: String,
    sku_id: Int,
    goods_code: String,
    batch_no: String,
    produced_on: String,
    expires_on: String,
    qty: Float,
  )
}

pub type ShipmentLine {
  ShipmentLine(
    line_no: Int,
    bill_detail_id: Int,
    sku_id: Int,
    goods_code: String,
    goods_name: String,
    batch_no: String,
    produced_on: String,
    expires_on: String,
    location_code: String,
    qty: Float,
  )
}

pub type Shipment {
  Shipment(
    id: String,
    bill_id: Int,
    warehouse_id: Int,
    created_at: String,
    lines: List(ShipmentLine),
  )
}

pub type BatchLabel {
  BatchLabel(
    batch_code: String,
    bill_id: Int,
    bill_code: String,
    bill_detail_id: Int,
    bill_type: Int,
    status: Int,
    warehouse_id: Int,
    warehouse_name: String,
    warehouse_found: Bool,
    operational: Bool,
    sku_id: Int,
    goods_code: String,
    goods_name: String,
    barcode: String,
    base_qty: Float,
    base_unit: String,
  )
}

pub type Putaway {
  Putaway(
    id: String,
    batch_code: String,
    bill_id: Int,
    bill_detail_id: Int,
    warehouse_id: Int,
    sku_id: Int,
    goods_code: String,
    goods_name: String,
    location_code: String,
    produced_on: String,
    expires_on: String,
    qty: Float,
    created_at: String,
  )
}

pub type OrderStamp {
  OrderStamp(
    status: Int,
    total_qty: Float,
    total_money: Float,
    bill_date: String,
    detail_synced_at: String,
  )
}

pub type SyncRun {
  SyncRun(
    id: String,
    entity: String,
    trigger: String,
    started_at: String,
    finished_at: String,
    status: String,
    fetched: Int,
    upserted: Int,
    error_message: String,
  )
}

pub type ProductPage {
  ProductPage(items: List(GoodsItem), total: Int)
}

pub type OrderPage {
  OrderPage(items: List(SalesOrderSummary), total: Int)
}

pub type StoredWarehouse {
  StoredWarehouse(warehouse: QlWarehouse, operational: Bool)
}

pub fn upsert_warehouse(
  conn: Conn,
  warehouse: QlWarehouse,
  operational: Bool,
  synced_at: String,
) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO warehouses (ql_id, code, name, easycode, operational, synced_at)
     VALUES (?, ?, ?, ?, ?, ?)
     ON CONFLICT (ql_id) DO UPDATE SET
       code = EXCLUDED.code,
       name = EXCLUDED.name,
       easycode = EXCLUDED.easycode,
       operational = EXCLUDED.operational,
       synced_at = EXCLUDED.synced_at",
    [
      IntParam(warehouse.id),
      TextParam(warehouse.code),
      TextParam(warehouse.fullname),
      TextParam(warehouse.easycode),
      IntParam(bool_bit(operational)),
      TextParam(synced_at),
    ],
  )
}

pub fn list_warehouses(conn: Conn) -> Result(List(StoredWarehouse), DbError) {
  database.query_on(
    conn,
    "SELECT ql_id, code, name, easycode, operational FROM warehouses ORDER BY code",
    [],
    {
      use id <- decode.field(0, json_codec.int())
      use code <- decode.field(1, decode.string)
      use name <- decode.field(2, decode.string)
      use easycode <- decode.field(3, decode.string)
      use operational <- decode.field(4, json_codec.int())
      decode.success(StoredWarehouse(
        warehouse: QlWarehouse(id: id, code: code, fullname: name, easycode: easycode),
        operational: operational != 0,
      ))
    },
  )
}

pub fn get_warehouse(
  conn: Conn,
  warehouse_id: Int,
) -> Result(Option(StoredWarehouse), DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT ql_id, code, name, easycode, operational FROM warehouses WHERE ql_id = ?",
    [IntParam(warehouse_id)],
    stored_warehouse_decoder(),
  ))
  case rows {
    [] -> Ok(None)
    [row, ..] -> Ok(Some(row))
  }
}

pub fn list_locations(
  conn: Conn,
  warehouse_id: Int,
) -> Result(List(Location), DbError) {
  database.query_on(
    conn,
    "SELECT warehouse_id, area, aisle, level_code, bin_code, code, barcode,
            enabled, created_at, updated_at
     FROM locations
     WHERE warehouse_id = ?
     ORDER BY code",
    [IntParam(warehouse_id)],
    location_decoder(),
  )
}

pub fn find_location(
  conn: Conn,
  warehouse_id: Int,
  code: String,
) -> Result(Option(Location), DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT warehouse_id, area, aisle, level_code, bin_code, code, barcode,
            enabled, created_at, updated_at
     FROM locations
     WHERE warehouse_id = ? AND code = ?",
    [IntParam(warehouse_id), TextParam(code)],
    location_decoder(),
  ))
  case rows {
    [] -> Ok(None)
    [row, ..] -> Ok(Some(row))
  }
}

pub fn insert_location(conn: Conn, row: Location) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO locations (
       warehouse_id, area, aisle, level_code, bin_code, code, barcode,
       enabled, created_at, updated_at
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
    [
      IntParam(row.warehouse_id),
      TextParam(row.area),
      TextParam(row.aisle),
      TextParam(row.level),
      TextParam(row.bin),
      TextParam(row.code),
      TextParam(row.barcode),
      IntParam(bool_bit(row.enabled)),
      TextParam(row.created_at),
      TextParam(row.updated_at),
    ],
  )
}

pub fn set_location_enabled(
  conn: Conn,
  warehouse_id: Int,
  code: String,
  enabled: Bool,
  updated_at: String,
) -> Result(Bool, DbError) {
  use existing <- result.try(find_location(conn, warehouse_id, code))
  case existing {
    None -> Ok(False)
    Some(_) -> {
      use _ <- result.try(database.exec_on(
        conn,
        "UPDATE locations SET enabled = ?, updated_at = ?
         WHERE warehouse_id = ? AND code = ?",
        [
          IntParam(bool_bit(enabled)),
          TextParam(updated_at),
          IntParam(warehouse_id),
          TextParam(code),
        ],
      ))
      Ok(True)
    }
  }
}

pub fn sum_received_qty(
  conn: Conn,
  bill_id: Int,
  bill_detail_id: Int,
) -> Result(Float, DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT COALESCE(SUM(l.received_qty), 0)
     FROM inbound_receipt_lines l
     JOIN inbound_receipts r ON r.id = l.receipt_id
     WHERE r.bill_id = ? AND l.bill_detail_id = ?",
    [IntParam(bill_id), IntParam(bill_detail_id)],
    {
      use qty <- decode.field(0, json_codec.number())
      decode.success(qty)
    },
  ))
  case rows {
    [qty, ..] -> Ok(qty)
    [] -> Ok(0.0)
  }
}

pub fn insert_receipt(conn: Conn, receipt: Receipt) -> Result(Nil, DbError) {
  use _ <- result.try(database.exec_on(
    conn,
    "INSERT INTO inbound_receipts (id, bill_id, warehouse_id, created_at)
     VALUES (?, ?, ?, ?)",
    [
      TextParam(receipt.id),
      IntParam(receipt.bill_id),
      IntParam(receipt.warehouse_id),
      TextParam(receipt.created_at),
    ],
  ))
  list.try_fold(receipt.lines, Nil, fn(_, line) {
    database.exec_on(
      conn,
      "INSERT INTO inbound_receipt_lines (
         receipt_id, line_no, bill_detail_id, sku_id, goods_code, goods_name,
         barcode, received_qty, batch_no, produced_on, expires_on, location_code
       ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
      [
        TextParam(receipt.id),
        IntParam(line.line_no),
        IntParam(line.bill_detail_id),
        IntParam(line.sku_id),
        TextParam(line.goods_code),
        TextParam(line.goods_name),
        TextParam(line.barcode),
        FloatParam(line.received_qty),
        TextParam(line.batch_no),
        TextParam(line.produced_on),
        TextParam(line.expires_on),
        TextParam(line.location_code),
      ],
    )
  })
}

pub fn add_inventory(
  conn: Conn,
  warehouse_id: Int,
  line: ReceiptLine,
  updated_at: String,
) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO inventory_batches (
       warehouse_id, location_code, sku_id, goods_code, batch_no,
       produced_on, expires_on, qty, updated_at
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT (
       warehouse_id, location_code, sku_id, batch_no, produced_on, expires_on
     ) DO UPDATE SET
       qty = inventory_batches.qty + EXCLUDED.qty,
       goods_code = EXCLUDED.goods_code,
       updated_at = EXCLUDED.updated_at",
    [
      IntParam(warehouse_id),
      TextParam(line.location_code),
      IntParam(line.sku_id),
      TextParam(line.goods_code),
      TextParam(line.batch_no),
      TextParam(line.produced_on),
      TextParam(line.expires_on),
      FloatParam(line.received_qty),
      TextParam(updated_at),
    ],
  )
}

pub fn list_receipts(conn: Conn, bill_id: Int) -> Result(List(Receipt), DbError) {
  use headers <- result.try(database.query_on(
    conn,
    "SELECT id, bill_id, warehouse_id, created_at
     FROM inbound_receipts WHERE bill_id = ? ORDER BY created_at, id",
    [IntParam(bill_id)],
    {
      use id <- decode.field(0, decode.string)
      use bill_id <- decode.field(1, json_codec.int())
      use warehouse_id <- decode.field(2, json_codec.int())
      use created_at <- decode.field(3, decode.string)
      decode.success(#(id, bill_id, warehouse_id, created_at))
    },
  ))
  list.try_map(headers, fn(header) {
    let #(id, bill_id, warehouse_id, created_at) = header
    use lines <- result.try(database.query_on(
      conn,
      "SELECT line_no, bill_detail_id, sku_id, goods_code, goods_name, barcode,
              received_qty, batch_no, produced_on, expires_on, location_code
       FROM inbound_receipt_lines WHERE receipt_id = ? ORDER BY line_no",
      [TextParam(id)],
      receipt_line_decoder(),
    ))
    Ok(Receipt(
      id: id,
      bill_id: bill_id,
      warehouse_id: warehouse_id,
      created_at: created_at,
      lines: lines,
    ))
  })
}

pub fn list_inventory(
  conn: Conn,
  warehouse_id: Int,
) -> Result(List(InventoryBatch), DbError) {
  database.query_on(
    conn,
    "SELECT warehouse_id, location_code, sku_id, goods_code, batch_no,
            produced_on, expires_on, qty
     FROM inventory_batches
     WHERE warehouse_id = ? AND qty > 0
     ORDER BY expires_on, location_code, goods_code, batch_no",
    [IntParam(warehouse_id)],
    {
      use warehouse_id <- decode.field(0, json_codec.int())
      use location_code <- decode.field(1, decode.string)
      use sku_id <- decode.field(2, json_codec.int())
      use goods_code <- decode.field(3, decode.string)
      use batch_no <- decode.field(4, decode.string)
      use produced_on <- decode.field(5, decode.string)
      use expires_on <- decode.field(6, decode.string)
      use qty <- decode.field(7, json_codec.number())
      decode.success(InventoryBatch(
        warehouse_id: warehouse_id,
        location_code: location_code,
        sku_id: sku_id,
        goods_code: goods_code,
        batch_no: batch_no,
        produced_on: produced_on,
        expires_on: expires_on,
        qty: qty,
      ))
    },
  )
}

pub fn list_batches_for_sku(
  conn: Conn,
  warehouse_id: Int,
  sku_id: Int,
  goods_code: String,
) -> Result(List(InventoryBatch), DbError) {
  database.query_on(
    conn,
    "SELECT warehouse_id, location_code, sku_id, goods_code, batch_no,
            produced_on, expires_on, qty
     FROM inventory_batches
     WHERE warehouse_id = ? AND qty > 0
       AND ((? <> 0 AND sku_id = ?) OR (? = 0 AND goods_code = ?))
     ORDER BY expires_on, location_code, batch_no",
    [
      IntParam(warehouse_id),
      IntParam(sku_id),
      IntParam(sku_id),
      IntParam(sku_id),
      TextParam(goods_code),
    ],
    inventory_batch_decoder(),
  )
}

pub fn sum_shipped_qty(
  conn: Conn,
  bill_id: Int,
  bill_detail_id: Int,
) -> Result(Float, DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT COALESCE(SUM(l.qty), 0)
     FROM outbound_shipment_lines l
     JOIN outbound_shipments s ON s.id = l.shipment_id
     WHERE s.bill_id = ? AND l.bill_detail_id = ?",
    [IntParam(bill_id), IntParam(bill_detail_id)],
    {
      use qty <- decode.field(0, json_codec.number())
      decode.success(qty)
    },
  ))
  case rows {
    [qty, ..] -> Ok(qty)
    [] -> Ok(0.0)
  }
}

pub fn deduct_inventory(
  conn: Conn,
  batch: InventoryBatch,
  qty: Float,
  updated_at: String,
) -> Result(Nil, DbError) {
  use before <- result.try(batch_qty(conn, batch))
  case before +. 0.0001 >=. qty {
    False -> Error(DbError("库存不足"))
    True -> {
      use _ <- result.try(database.exec_on(
        conn,
        "UPDATE inventory_batches
         SET qty = qty - ?, updated_at = ?
         WHERE warehouse_id = ? AND location_code = ? AND sku_id = ?
           AND batch_no = ? AND produced_on = ? AND expires_on = ?
           AND qty + 0.0001 >= ?",
        [
          FloatParam(qty),
          TextParam(updated_at),
          IntParam(batch.warehouse_id),
          TextParam(batch.location_code),
          IntParam(batch.sku_id),
          TextParam(batch.batch_no),
          TextParam(batch.produced_on),
          TextParam(batch.expires_on),
          FloatParam(qty),
        ],
      ))
      use after <- result.try(batch_qty(conn, batch))
      let expected = before -. qty
      case after <=. expected +. 0.0001 && after >=. expected -. 0.0001 {
        True -> Ok(Nil)
        False -> Error(DbError("库存不足"))
      }
    }
  }
}

fn batch_qty(conn: Conn, batch: InventoryBatch) -> Result(Float, DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT qty FROM inventory_batches
     WHERE warehouse_id = ? AND location_code = ? AND sku_id = ?
       AND batch_no = ? AND produced_on = ? AND expires_on = ?",
    [
      IntParam(batch.warehouse_id),
      TextParam(batch.location_code),
      IntParam(batch.sku_id),
      TextParam(batch.batch_no),
      TextParam(batch.produced_on),
      TextParam(batch.expires_on),
    ],
    {
      use qty <- decode.field(0, json_codec.number())
      decode.success(qty)
    },
  ))
  case rows {
    [qty, ..] -> Ok(qty)
    [] -> Ok(0.0)
  }
}

pub fn insert_shipment(conn: Conn, shipment: Shipment) -> Result(Nil, DbError) {
  use _ <- result.try(database.exec_on(
    conn,
    "INSERT INTO outbound_shipments (id, bill_id, warehouse_id, created_at)
     VALUES (?, ?, ?, ?)",
    [
      TextParam(shipment.id),
      IntParam(shipment.bill_id),
      IntParam(shipment.warehouse_id),
      TextParam(shipment.created_at),
    ],
  ))
  list.try_fold(shipment.lines, Nil, fn(_, line) {
    database.exec_on(
      conn,
      "INSERT INTO outbound_shipment_lines (
         shipment_id, line_no, bill_detail_id, sku_id, goods_code, goods_name,
         batch_no, produced_on, expires_on, location_code, qty
       ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
      [
        TextParam(shipment.id),
        IntParam(line.line_no),
        IntParam(line.bill_detail_id),
        IntParam(line.sku_id),
        TextParam(line.goods_code),
        TextParam(line.goods_name),
        TextParam(line.batch_no),
        TextParam(line.produced_on),
        TextParam(line.expires_on),
        TextParam(line.location_code),
        FloatParam(line.qty),
      ],
    )
  })
}

pub fn list_shipments(conn: Conn, bill_id: Int) -> Result(List(Shipment), DbError) {
  use headers <- result.try(database.query_on(
    conn,
    "SELECT id, bill_id, warehouse_id, created_at
     FROM outbound_shipments WHERE bill_id = ? ORDER BY created_at, id",
    [IntParam(bill_id)],
    {
      use id <- decode.field(0, decode.string)
      use bill_id <- decode.field(1, json_codec.int())
      use warehouse_id <- decode.field(2, json_codec.int())
      use created_at <- decode.field(3, decode.string)
      decode.success(#(id, bill_id, warehouse_id, created_at))
    },
  ))
  list.try_map(headers, fn(header) {
    let #(id, bill_id, warehouse_id, created_at) = header
    use lines <- result.try(database.query_on(
      conn,
      "SELECT line_no, bill_detail_id, sku_id, goods_code, goods_name, batch_no,
              produced_on, expires_on, location_code, qty
       FROM outbound_shipment_lines WHERE shipment_id = ? ORDER BY line_no",
      [TextParam(id)],
      shipment_line_decoder(),
    ))
    Ok(Shipment(
      id: id,
      bill_id: bill_id,
      warehouse_id: warehouse_id,
      created_at: created_at,
      lines: lines,
    ))
  })
}

fn inventory_batch_decoder() -> decode.Decoder(InventoryBatch) {
  use warehouse_id <- decode.field(0, json_codec.int())
  use location_code <- decode.field(1, decode.string)
  use sku_id <- decode.field(2, json_codec.int())
  use goods_code <- decode.field(3, decode.string)
  use batch_no <- decode.field(4, decode.string)
  use produced_on <- decode.field(5, decode.string)
  use expires_on <- decode.field(6, decode.string)
  use qty <- decode.field(7, json_codec.number())
  decode.success(InventoryBatch(
    warehouse_id: warehouse_id,
    location_code: location_code,
    sku_id: sku_id,
    goods_code: goods_code,
    batch_no: batch_no,
    produced_on: produced_on,
    expires_on: expires_on,
    qty: qty,
  ))
}

fn shipment_line_decoder() -> decode.Decoder(ShipmentLine) {
  use line_no <- decode.field(0, json_codec.int())
  use bill_detail_id <- decode.field(1, json_codec.int())
  use sku_id <- decode.field(2, json_codec.int())
  use goods_code <- decode.field(3, decode.string)
  use goods_name <- decode.field(4, decode.string)
  use batch_no <- decode.field(5, decode.string)
  use produced_on <- decode.field(6, decode.string)
  use expires_on <- decode.field(7, decode.string)
  use location_code <- decode.field(8, decode.string)
  use qty <- decode.field(9, json_codec.number())
  decode.success(ShipmentLine(
    line_no: line_no,
    bill_detail_id: bill_detail_id,
    sku_id: sku_id,
    goods_code: goods_code,
    goods_name: goods_name,
    batch_no: batch_no,
    produced_on: produced_on,
    expires_on: expires_on,
    location_code: location_code,
    qty: qty,
  ))
}

fn receipt_line_decoder() -> decode.Decoder(ReceiptLine) {
  use line_no <- decode.field(0, json_codec.int())
  use bill_detail_id <- decode.field(1, json_codec.int())
  use sku_id <- decode.field(2, json_codec.int())
  use goods_code <- decode.field(3, decode.string)
  use goods_name <- decode.field(4, decode.string)
  use barcode <- decode.field(5, decode.string)
  use received_qty <- decode.field(6, json_codec.number())
  use batch_no <- decode.field(7, decode.string)
  use produced_on <- decode.field(8, decode.string)
  use expires_on <- decode.field(9, decode.string)
  use location_code <- decode.field(10, decode.string)
  decode.success(ReceiptLine(
    line_no: line_no,
    bill_detail_id: bill_detail_id,
    sku_id: sku_id,
    goods_code: goods_code,
    goods_name: goods_name,
    barcode: barcode,
    received_qty: received_qty,
    batch_no: batch_no,
    produced_on: produced_on,
    expires_on: expires_on,
    location_code: location_code,
  ))
}

fn stored_warehouse_decoder() -> decode.Decoder(StoredWarehouse) {
  use id <- decode.field(0, json_codec.int())
  use code <- decode.field(1, decode.string)
  use name <- decode.field(2, decode.string)
  use easycode <- decode.field(3, decode.string)
  use operational <- decode.field(4, json_codec.int())
  decode.success(StoredWarehouse(
    warehouse: QlWarehouse(id: id, code: code, fullname: name, easycode: easycode),
    operational: operational != 0,
  ))
}

fn location_decoder() -> decode.Decoder(Location) {
  use warehouse_id <- decode.field(0, json_codec.int())
  use area <- decode.field(1, decode.string)
  use aisle <- decode.field(2, decode.string)
  use level <- decode.field(3, decode.string)
  use bin <- decode.field(4, decode.string)
  use code <- decode.field(5, decode.string)
  use barcode <- decode.field(6, decode.string)
  use enabled <- decode.field(7, json_codec.int())
  use created_at <- decode.field(8, decode.string)
  use updated_at <- decode.field(9, decode.string)
  decode.success(Location(
    warehouse_id: warehouse_id,
    area: area,
    aisle: aisle,
    level: level,
    bin: bin,
    code: code,
    barcode: barcode,
    enabled: enabled != 0,
    created_at: created_at,
    updated_at: updated_at,
  ))
}

pub fn delete_documents_outside_warehouses(
  conn: Conn,
  warehouse_ids: List(Int),
) -> Result(Nil, DbError) {
  case warehouse_ids {
    [] -> Ok(Nil)
    _ -> {
      let marks = string.join(list.map(warehouse_ids, fn(_) { "?" }), ",")
      let params = list.map(warehouse_ids, IntParam)
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM outbound_shipment_lines WHERE shipment_id IN (
           SELECT id FROM outbound_shipments WHERE bill_id IN (
             SELECT bill_id FROM outbound_orders WHERE warehouse_id NOT IN ("
          <> marks
          <> ")))",
        params,
      ))
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM outbound_shipments WHERE bill_id IN (
           SELECT bill_id FROM outbound_orders WHERE warehouse_id NOT IN ("
          <> marks
          <> "))",
        params,
      ))
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM outbound_order_items WHERE bill_id IN (
           SELECT bill_id FROM outbound_orders WHERE warehouse_id NOT IN ("
          <> marks
          <> "))",
        params,
      ))
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM outbound_orders WHERE warehouse_id NOT IN (" <> marks <> ")",
        params,
      ))
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM inbound_receipt_lines WHERE receipt_id IN (
           SELECT id FROM inbound_receipts WHERE bill_id IN (
             SELECT bill_id FROM inbound_orders WHERE warehouse_id NOT IN ("
          <> marks
          <> ")))",
        params,
      ))
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM inbound_receipts WHERE bill_id IN (
           SELECT bill_id FROM inbound_orders WHERE warehouse_id NOT IN ("
          <> marks
          <> "))",
        params,
      ))
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM inbound_order_items WHERE bill_id IN (
           SELECT bill_id FROM inbound_orders WHERE warehouse_id NOT IN ("
          <> marks
          <> "))",
        params,
      ))
      database.exec_on(
        conn,
        "DELETE FROM inbound_orders WHERE warehouse_id NOT IN (" <> marks <> ")",
        params,
      )
    }
  }
}

fn bool_bit(value: Bool) -> Int {
  case value {
    True -> 1
    False -> 0
  }
}

pub fn upsert_product(
  conn: Conn,
  item: GoodsItem,
  synced_at: String,
) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO products (
       ql_id, code, name, barcode, spec, unit, shelf_life, shelf_life_days,
       position_code, disabled, is_category, source_modified_at, source_created_at, synced_at
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT (ql_id) DO UPDATE SET
       code = EXCLUDED.code,
       name = EXCLUDED.name,
       barcode = EXCLUDED.barcode,
       spec = EXCLUDED.spec,
       unit = EXCLUDED.unit,
       shelf_life = EXCLUDED.shelf_life,
       shelf_life_days = EXCLUDED.shelf_life_days,
       position_code = EXCLUDED.position_code,
       disabled = EXCLUDED.disabled,
       is_category = EXCLUDED.is_category,
       source_modified_at = EXCLUDED.source_modified_at,
       source_created_at = EXCLUDED.source_created_at,
       synced_at = EXCLUDED.synced_at",
    [
      IntParam(item.id),
      TextParam(item.code),
      TextParam(item.name),
      TextParam(item.barcode),
      TextParam(item.spec),
      TextParam(item.unit),
      TextParam(item.shelf_life),
      IntParam(item.shelf_life_days),
      TextParam(item.position_code),
      IntParam(bool_int(item.disabled)),
      IntParam(bool_int(item.is_category)),
      TextParam(item.modified_at),
      TextParam(item.created_at),
      TextParam(synced_at),
    ],
  )
}

pub fn list_products(
  conn: Conn,
  keyword: String,
  limit: Int,
  offset: Int,
) -> Result(ProductPage, DbError) {
  let pattern = like_pattern(keyword)
  let filter =
    "is_category = 0 AND (? = '' OR code LIKE ? OR name LIKE ? OR barcode LIKE ?)"
  let params = [
    TextParam(keyword),
    TextParam(pattern),
    TextParam(pattern),
    TextParam(pattern),
  ]
  use totals <- result.try(database.query_on(
    conn,
    "SELECT COUNT(*) FROM products WHERE " <> filter,
    params,
    {
      use total <- decode.field(0, json_codec.int())
      decode.success(total)
    },
  ))
  let total = case totals {
    [value, ..] -> value
    [] -> 0
  }
  use items <- result.try(database.query_on(
    conn,
    "SELECT ql_id, code, name, barcode, spec, unit, shelf_life, shelf_life_days,
            position_code, disabled, is_category, source_modified_at, source_created_at
     FROM products
     WHERE " <> filter <> "
     ORDER BY code
     LIMIT ? OFFSET ?",
    list.append(params, [IntParam(limit), IntParam(offset)]),
    product_decoder(),
  ))
  Ok(ProductPage(items: items, total: total))
}

pub fn upsert_order_summary(
  conn: Conn,
  order: SalesOrderSummary,
  synced_at: String,
) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO outbound_orders (
       bill_id, bill_code, customer_name, contact_person, phone,
       warehouse_name, warehouse_id, bill_date, status, status_name,
       total_qty, total_money, remark, detail_synced_at, synced_at
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '', ?)
     ON CONFLICT (bill_id) DO UPDATE SET
       bill_code = EXCLUDED.bill_code,
       customer_name = EXCLUDED.customer_name,
       contact_person = EXCLUDED.contact_person,
       phone = EXCLUDED.phone,
       warehouse_name = EXCLUDED.warehouse_name,
       warehouse_id = EXCLUDED.warehouse_id,
       bill_date = EXCLUDED.bill_date,
       status = EXCLUDED.status,
       status_name = EXCLUDED.status_name,
       total_qty = EXCLUDED.total_qty,
       total_money = EXCLUDED.total_money,
       remark = EXCLUDED.remark,
       synced_at = EXCLUDED.synced_at",
    order_params(order, synced_at),
  )
}

pub fn save_order_detail(
  conn: Conn,
  order: SalesOrder,
  synced_at: String,
) -> Result(Nil, DbError) {
  let summary =
    SalesOrderSummary(
      bill_id: order.bill_id,
      bill_code: order.bill_code,
      customer_name: order.customer_name,
      contact_person: order.contact_person,
      phone: order.phone,
      warehouse_name: order.warehouse_name,
      warehouse_id: order.warehouse_id,
      bill_date: order.bill_date,
      status: order.status,
      status_name: order.status_name,
      total_qty: order.total_qty,
      total_money: order.total_money,
      remark: order.remark,
    )
  use _ <- result.try(database.exec_on(
    conn,
    "INSERT INTO outbound_orders (
       bill_id, bill_code, customer_name, contact_person, phone,
       warehouse_name, warehouse_id, bill_date, status, status_name,
       total_qty, total_money, remark, detail_synced_at, synced_at
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT (bill_id) DO UPDATE SET
       bill_code = EXCLUDED.bill_code,
       customer_name = EXCLUDED.customer_name,
       contact_person = EXCLUDED.contact_person,
       phone = EXCLUDED.phone,
       warehouse_name = EXCLUDED.warehouse_name,
       warehouse_id = EXCLUDED.warehouse_id,
       bill_date = EXCLUDED.bill_date,
       status = EXCLUDED.status,
       status_name = EXCLUDED.status_name,
       total_qty = EXCLUDED.total_qty,
       total_money = EXCLUDED.total_money,
       remark = EXCLUDED.remark,
       detail_synced_at = EXCLUDED.detail_synced_at,
       synced_at = EXCLUDED.synced_at",
    list.append(order_params(summary, synced_at), [TextParam(synced_at)]),
  ))
  use _ <- result.try(database.exec_on(
    conn,
    "DELETE FROM outbound_order_items WHERE bill_id = ?",
    [IntParam(order.bill_id)],
  ))
  list.index_map(order.items, fn(item, index) { #(index + 1, item) })
  |> list.try_fold(Nil, fn(_, pair) {
    let #(line_no, item) = pair
    insert_item(conn, order.bill_id, line_no, item)
  })
}

pub fn get_order_stamp(
  conn: Conn,
  bill_id: Int,
) -> Result(Option(OrderStamp), DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT status, total_qty, total_money, bill_date, detail_synced_at
     FROM outbound_orders WHERE bill_id = ?",
    [IntParam(bill_id)],
    {
      use status <- decode.field(0, json_codec.int())
      use total_qty <- decode.field(1, json_codec.number())
      use total_money <- decode.field(2, json_codec.number())
      use bill_date <- decode.field(3, decode.string)
      use detail_synced_at <- decode.field(4, decode.string)
      decode.success(OrderStamp(
        status: status,
        total_qty: total_qty,
        total_money: total_money,
        bill_date: bill_date,
        detail_synced_at: detail_synced_at,
      ))
    },
  ))
  case rows {
    [] -> Ok(None)
    [stamp, ..] -> Ok(Some(stamp))
  }
}

pub fn list_pending_bill_ids(
  conn: Conn,
  begin_date: String,
  end_date: String,
  warehouse_ids: List(Int),
) -> Result(List(Int), DbError) {
  list_pending_ids(conn, "outbound_orders", begin_date, end_date, warehouse_ids)
}

const shipped_out_filter = "EXISTS (
       SELECT 1 FROM outbound_order_items i WHERE i.bill_id = outbound_orders.bill_id
     ) AND NOT EXISTS (
       SELECT 1 FROM outbound_order_items i
       WHERE i.bill_id = outbound_orders.bill_id
         AND i.base_qty > COALESCE((
           SELECT SUM(l.qty)
           FROM outbound_shipment_lines l
           JOIN outbound_shipments s ON s.id = l.shipment_id
           WHERE s.bill_id = i.bill_id AND l.bill_detail_id = i.bill_detail_id
         ), 0) + 0.0001
     )"

pub fn list_orders(
  conn: Conn,
  scope: rbac.Scope,
  status: Option(Int),
  limit: Int,
  offset: Int,
) -> Result(OrderPage, DbError) {
  let #(status_flag, status_value) = case status {
    Some(value) -> #(value, value)
    None -> #(-1, -1)
  }
  let #(scope_sql, scope_params) = rbac.sql_filter(scope, "warehouse_id")
  let filter = case status {
    Some(20) -> "status = 20 AND ? = ? AND NOT (" <> shipped_out_filter <> ")"
    _ -> "(? < 0 OR status = ?)"
  }
  let filter = filter <> " AND " <> scope_sql
  let params = [IntParam(status_flag), IntParam(status_value), ..scope_params]
  use totals <- result.try(database.query_on(
    conn,
    "SELECT COUNT(*) FROM outbound_orders WHERE " <> filter,
    params,
    {
      use total <- decode.field(0, json_codec.int())
      decode.success(total)
    },
  ))
  let total = case totals {
    [value, ..] -> value
    [] -> 0
  }
  use items <- result.try(database.query_on(
    conn,
    "SELECT bill_id, bill_code, customer_name, contact_person, phone,
            warehouse_name, warehouse_id, bill_date, status, status_name,
            total_qty, total_money, remark
     FROM outbound_orders
     WHERE " <> filter <> "
     ORDER BY bill_date DESC, bill_id DESC
     LIMIT ? OFFSET ?",
    list.append(params, [IntParam(limit), IntParam(offset)]),
    summary_decoder(),
  ))
  Ok(OrderPage(items: items, total: total))
}

/// 当前仓库里还能组波次的待发货订单：未发完，且不在未完成的波次中。
pub fn list_wave_candidates(
  conn: Conn,
  scope: rbac.Scope,
  warehouse_id: Int,
  limit: Int,
  offset: Int,
) -> Result(OrderPage, DbError) {
  let #(scope_sql, scope_params) = rbac.sql_filter(scope, "warehouse_id")
  let filter =
    "status = 20 AND warehouse_id = ? AND NOT ("
    <> shipped_out_filter
    <> ") AND NOT EXISTS (
         SELECT 1 FROM wave_orders wo
         JOIN waves w ON w.id = wo.wave_id
         WHERE wo.bill_id = outbound_orders.bill_id AND w.status = 'open'
       ) AND "
    <> scope_sql
  let params = [IntParam(warehouse_id), ..scope_params]
  use totals <- result.try(database.query_on(
    conn,
    "SELECT COUNT(*) FROM outbound_orders WHERE " <> filter,
    params,
    {
      use total <- decode.field(0, json_codec.int())
      decode.success(total)
    },
  ))
  let total = case totals {
    [value, ..] -> value
    [] -> 0
  }
  use items <- result.try(database.query_on(
    conn,
    "SELECT bill_id, bill_code, customer_name, contact_person, phone,
            warehouse_name, warehouse_id, bill_date, status, status_name,
            total_qty, total_money, remark
     FROM outbound_orders
     WHERE " <> filter <> "
     ORDER BY bill_date, bill_id
     LIMIT ? OFFSET ?",
    list.append(params, [IntParam(limit), IntParam(offset)]),
    summary_decoder(),
  ))
  Ok(OrderPage(items: items, total: total))
}

pub fn order_shipped_out(conn: Conn, bill_id: Int) -> Result(Bool, DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT COUNT(*) FROM outbound_orders
     WHERE bill_id = ? AND " <> shipped_out_filter,
    [IntParam(bill_id)],
    {
      use count <- decode.field(0, json_codec.int())
      decode.success(count)
    },
  ))
  case rows {
    [count, ..] -> Ok(count > 0)
    [] -> Ok(False)
  }
}

pub fn get_order(conn: Conn, bill_id: Int) -> Result(Option(SalesOrder), DbError) {
  use headers <- result.try(database.query_on(
    conn,
    "SELECT bill_id, bill_code, customer_name, contact_person, phone,
            warehouse_name, warehouse_id, bill_date, status, status_name,
            total_qty, total_money, remark
     FROM outbound_orders WHERE bill_id = ?",
    [IntParam(bill_id)],
    summary_decoder(),
  ))
  case headers {
    [] -> Ok(None)
    [summary, ..] -> {
      use items <- result.try(database.query_on(
        conn,
        "SELECT bill_detail_id, sku_id, goods_code, goods_name, barcode, spec,
                package_qty, package_unit, base_qty, base_unit, price, total_money,
                position_code, area_code
         FROM outbound_order_items
         WHERE bill_id = ?
         ORDER BY line_no",
        [IntParam(bill_id)],
        item_decoder(),
      ))
      Ok(Some(SalesOrder(
        bill_id: summary.bill_id,
        bill_code: summary.bill_code,
        customer_name: summary.customer_name,
        contact_person: summary.contact_person,
        phone: summary.phone,
        warehouse_name: summary.warehouse_name,
        warehouse_id: summary.warehouse_id,
        bill_date: summary.bill_date,
        status: summary.status,
        status_name: summary.status_name,
        total_qty: summary.total_qty,
        total_money: summary.total_money,
        remark: summary.remark,
        items: items,
      )))
    }
  }
}

pub fn upsert_inbound_summary(
  conn: Conn,
  order: PurchaseOrderSummary,
  bill_type: Int,
  synced_at: String,
) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO inbound_orders (
       bill_id, bill_code, supplier_name, contact_person, phone,
       warehouse_name, warehouse_id, bill_date, status, status_name,
       total_qty, total_money, remark, detail_synced_at, synced_at, bill_type
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '', ?, ?)
     ON CONFLICT (bill_id) DO UPDATE SET
       bill_code = EXCLUDED.bill_code,
       supplier_name = EXCLUDED.supplier_name,
       contact_person = EXCLUDED.contact_person,
       phone = EXCLUDED.phone,
       warehouse_name = EXCLUDED.warehouse_name,
       warehouse_id = EXCLUDED.warehouse_id,
       bill_date = EXCLUDED.bill_date,
       status = EXCLUDED.status,
       status_name = EXCLUDED.status_name,
       total_qty = EXCLUDED.total_qty,
       total_money = EXCLUDED.total_money,
       remark = EXCLUDED.remark,
       synced_at = EXCLUDED.synced_at,
       bill_type = EXCLUDED.bill_type",
    list.append(inbound_params(order, synced_at), [IntParam(bill_type)]),
  )
}

pub fn save_inbound_detail(
  conn: Conn,
  order: PurchaseOrder,
  bill_type: Int,
  synced_at: String,
) -> Result(Nil, DbError) {
  let summary =
    PurchaseOrderSummary(
      bill_id: order.bill_id,
      bill_code: order.bill_code,
      supplier_name: order.supplier_name,
      contact_person: order.contact_person,
      phone: order.phone,
      warehouse_name: order.warehouse_name,
      warehouse_id: order.warehouse_id,
      bill_date: order.bill_date,
      status: order.status,
      status_name: order.status_name,
      total_qty: order.total_qty,
      total_money: order.total_money,
      remark: order.remark,
      ifred: 0,
    )
  use _ <- result.try(database.exec_on(
    conn,
    "INSERT INTO inbound_orders (
       bill_id, bill_code, supplier_name, contact_person, phone,
       warehouse_name, warehouse_id, bill_date, status, status_name,
       total_qty, total_money, remark, detail_synced_at, synced_at, bill_type
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     ON CONFLICT (bill_id) DO UPDATE SET
       bill_code = EXCLUDED.bill_code,
       supplier_name = EXCLUDED.supplier_name,
       contact_person = EXCLUDED.contact_person,
       phone = EXCLUDED.phone,
       warehouse_name = EXCLUDED.warehouse_name,
       warehouse_id = EXCLUDED.warehouse_id,
       bill_date = EXCLUDED.bill_date,
       status = EXCLUDED.status,
       status_name = EXCLUDED.status_name,
       total_qty = EXCLUDED.total_qty,
       total_money = EXCLUDED.total_money,
       remark = EXCLUDED.remark,
       detail_synced_at = EXCLUDED.detail_synced_at,
       synced_at = EXCLUDED.synced_at,
       bill_type = EXCLUDED.bill_type",
    list.append(inbound_params(summary, synced_at), [
      TextParam(synced_at),
      IntParam(bill_type),
    ]),
  ))
  use _ <- result.try(database.exec_on(
    conn,
    "DELETE FROM inbound_order_items WHERE bill_id = ?",
    [IntParam(order.bill_id)],
  ))
  use _ <- result.try(
    list.index_map(order.items, fn(item, index) { #(index + 1, item) })
    |> list.try_fold(Nil, fn(_, pair) {
      let #(line_no, item) = pair
      insert_inbound_item(conn, order.bill_id, line_no, item)
    }),
  )
  case bill_type == 501 {
    True -> ensure_batch_codes(conn, order.bill_id, order.items, synced_at)
    False -> Ok(Nil)
  }
}

pub fn list_batch_codes(
  conn: Conn,
  bill_id: Int,
) -> Result(List(#(Int, String)), DbError) {
  database.query_on(
    conn,
    "SELECT bill_detail_id, batch_code FROM inbound_batch_codes WHERE bill_id = ?",
    [IntParam(bill_id)],
    {
      use bill_detail_id <- decode.field(0, json_codec.int())
      use code <- decode.field(1, decode.string)
      decode.success(#(bill_detail_id, code))
    },
  )
}

pub fn find_batch_label(
  conn: Conn,
  batch_code: String,
) -> Result(Option(BatchLabel), DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT c.batch_code, c.bill_id, o.bill_code, c.bill_detail_id, o.bill_type, o.status,
            o.warehouse_id, o.warehouse_name,
            CASE WHEN w.ql_id IS NULL THEN 0 ELSE 1 END,
            CASE WHEN w.operational IS NULL THEN 0 ELSE w.operational END,
            i.sku_id, i.goods_code, i.goods_name, i.barcode, i.base_qty, i.base_unit
     FROM inbound_batch_codes c
     JOIN inbound_orders o ON o.bill_id = c.bill_id
     JOIN inbound_order_items i
       ON i.bill_id = c.bill_id AND i.bill_detail_id = c.bill_detail_id
     LEFT JOIN warehouses w ON w.ql_id = o.warehouse_id
     WHERE c.batch_code = ?",
    [TextParam(batch_code)],
    {
      use batch_code <- decode.field(0, decode.string)
      use bill_id <- decode.field(1, json_codec.int())
      use bill_code <- decode.field(2, decode.string)
      use bill_detail_id <- decode.field(3, json_codec.int())
      use bill_type <- decode.field(4, json_codec.int())
      use status <- decode.field(5, json_codec.int())
      use warehouse_id <- decode.field(6, json_codec.int())
      use warehouse_name <- decode.field(7, decode.string)
      use warehouse_found <- decode.field(8, json_codec.int())
      use operational <- decode.field(9, json_codec.int())
      use sku_id <- decode.field(10, json_codec.int())
      use goods_code <- decode.field(11, decode.string)
      use goods_name <- decode.field(12, decode.string)
      use barcode <- decode.field(13, decode.string)
      use base_qty <- decode.field(14, json_codec.number())
      use base_unit <- decode.field(15, decode.string)
      decode.success(BatchLabel(
        batch_code: batch_code,
        bill_id: bill_id,
        bill_code: bill_code,
        bill_detail_id: bill_detail_id,
        bill_type: bill_type,
        status: status,
        warehouse_id: warehouse_id,
        warehouse_name: warehouse_name,
        warehouse_found: warehouse_found != 0,
        operational: operational != 0,
        sku_id: sku_id,
        goods_code: goods_code,
        goods_name: goods_name,
        barcode: barcode,
        base_qty: base_qty,
        base_unit: base_unit,
      ))
    },
  ))
  case rows {
    [] -> Ok(None)
    [row, ..] -> Ok(Some(row))
  }
}

pub fn sum_putaway_qty(conn: Conn, batch_code: String) -> Result(Float, DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT COALESCE(SUM(qty), 0) FROM putaways WHERE batch_code = ?",
    [TextParam(batch_code)],
    {
      use qty <- decode.field(0, json_codec.number())
      decode.success(qty)
    },
  ))
  case rows {
    [qty, ..] -> Ok(qty)
    [] -> Ok(0.0)
  }
}

pub fn putaway_dates(
  conn: Conn,
  batch_code: String,
) -> Result(Option(#(String, String)), DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT produced_on, expires_on FROM putaways
     WHERE batch_code = ? ORDER BY created_at, id LIMIT 1",
    [TextParam(batch_code)],
    {
      use produced_on <- decode.field(0, decode.string)
      use expires_on <- decode.field(1, decode.string)
      decode.success(#(produced_on, expires_on))
    },
  ))
  case rows {
    [] -> Ok(None)
    [row, ..] -> Ok(Some(row))
  }
}

pub fn insert_putaway(conn: Conn, row: Putaway) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO putaways (
       id, batch_code, bill_id, bill_detail_id, warehouse_id, sku_id,
       goods_code, goods_name, location_code, produced_on, expires_on, qty, created_at
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
    [
      TextParam(row.id),
      TextParam(row.batch_code),
      IntParam(row.bill_id),
      IntParam(row.bill_detail_id),
      IntParam(row.warehouse_id),
      IntParam(row.sku_id),
      TextParam(row.goods_code),
      TextParam(row.goods_name),
      TextParam(row.location_code),
      TextParam(row.produced_on),
      TextParam(row.expires_on),
      FloatParam(row.qty),
      TextParam(row.created_at),
    ],
  )
}

fn ensure_batch_codes(
  conn: Conn,
  bill_id: Int,
  items: List(PurchaseOrderItem),
  created_at: String,
) -> Result(Nil, DbError) {
  list.try_fold(items, Nil, fn(_, item) {
    database.exec_on(
      conn,
      "INSERT INTO inbound_batch_codes (bill_id, bill_detail_id, batch_code, created_at)
       VALUES (?, ?, ?, ?)
       ON CONFLICT (bill_id, bill_detail_id) DO NOTHING",
      [
        IntParam(bill_id),
        IntParam(item.bill_detail_id),
        TextParam(batch_code.for_detail(item.bill_detail_id)),
        TextParam(created_at),
      ],
    )
  })
}

pub fn delete_posted_receipts_absent(
  conn: Conn,
  begin_date: String,
  end_date: String,
  warehouse_ids: List(Int),
  seen_ids: List(Int),
) -> Result(Nil, DbError) {
  case warehouse_ids {
    [] -> Ok(Nil)
    _ -> {
      let warehouse_marks = string.join(list.map(warehouse_ids, fn(_) { "?" }), ",")
      let seen_clause = case seen_ids {
        [] -> ""
        _ ->
          " AND bill_id NOT IN ("
          <> string.join(list.map(seen_ids, fn(_) { "?" }), ",")
          <> ")"
      }
      let params =
        list.append(
          [TextParam(begin_date), TextParam(end_date)],
          list.append(
            list.map(warehouse_ids, IntParam),
            list.map(seen_ids, IntParam),
          ),
        )
      let where_ids =
        "SELECT bill_id FROM inbound_orders
         WHERE bill_type = 501 AND status = 20
           AND substr(bill_date, 1, 10) >= ?
           AND substr(bill_date, 1, 10) <= ?
           AND warehouse_id IN ("
        <> warehouse_marks
        <> ")"
        <> seen_clause
        <> " AND bill_id NOT IN (SELECT bill_id FROM putaways)"
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM inbound_batch_codes WHERE bill_id IN (" <> where_ids <> ")",
        params,
      ))
      use _ <- result.try(database.exec_on(
        conn,
        "DELETE FROM inbound_order_items WHERE bill_id IN (" <> where_ids <> ")",
        params,
      ))
      database.exec_on(
        conn,
        "DELETE FROM inbound_orders WHERE bill_id IN (" <> where_ids <> ")",
        params,
      )
    }
  }
}

pub fn get_inbound_stamp(
  conn: Conn,
  bill_id: Int,
) -> Result(Option(OrderStamp), DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT status, total_qty, total_money, bill_date, detail_synced_at
     FROM inbound_orders WHERE bill_id = ?",
    [IntParam(bill_id)],
    stamp_decoder(),
  ))
  case rows {
    [] -> Ok(None)
    [stamp, ..] -> Ok(Some(stamp))
  }
}

pub fn list_pending_inbound_bill_ids(
  conn: Conn,
  begin_date: String,
  end_date: String,
  warehouse_ids: List(Int),
) -> Result(List(Int), DbError) {
  list_inbound_bill_ids(conn, 504, begin_date, end_date, warehouse_ids)
}

fn list_inbound_bill_ids(
  conn: Conn,
  bill_type: Int,
  begin_date: String,
  end_date: String,
  warehouse_ids: List(Int),
) -> Result(List(Int), DbError) {
  case warehouse_ids {
    [] -> Ok([])
    _ -> {
      let marks = string.join(list.map(warehouse_ids, fn(_) { "?" }), ",")
      database.query_on(
        conn,
        "SELECT bill_id FROM inbound_orders
         WHERE bill_type = ? AND status = 20
           AND substr(bill_date, 1, 10) >= ?
           AND substr(bill_date, 1, 10) <= ?
           AND warehouse_id IN ("
          <> marks
          <> ")",
        list.append(
          [IntParam(bill_type), TextParam(begin_date), TextParam(end_date)],
          list.map(warehouse_ids, IntParam),
        ),
        {
          use bill_id <- decode.field(0, json_codec.int())
          decode.success(bill_id)
        },
      )
    }
  }
}

fn list_pending_ids(
  conn: Conn,
  table: String,
  begin_date: String,
  end_date: String,
  warehouse_ids: List(Int),
) -> Result(List(Int), DbError) {
  case warehouse_ids {
    [] -> Ok([])
    _ -> {
      let marks = string.join(list.map(warehouse_ids, fn(_) { "?" }), ",")
      database.query_on(
        conn,
        "SELECT bill_id FROM "
          <> table
          <> "
         WHERE status = 20
           AND substr(bill_date, 1, 10) >= ?
           AND substr(bill_date, 1, 10) <= ?
           AND warehouse_id IN ("
          <> marks
          <> ")",
        list.append(
          [TextParam(begin_date), TextParam(end_date)],
          list.map(warehouse_ids, IntParam),
        ),
        {
          use bill_id <- decode.field(0, json_codec.int())
          decode.success(bill_id)
        },
      )
    }
  }
}

pub fn list_inbounds(
  conn: Conn,
  scope: rbac.Scope,
  bill_type: Int,
  status: Option(Int),
  limit: Int,
  offset: Int,
) -> Result(InboundPage, DbError) {
  let #(status_flag, status_value) = case status {
    Some(value) -> #(value, value)
    None -> #(-1, -1)
  }
  let #(scope_sql, scope_params) = rbac.sql_filter(scope, "warehouse_id")
  let filter = "bill_type = ? AND (? < 0 OR status = ?) AND " <> scope_sql
  let params = [
    IntParam(bill_type),
    IntParam(status_flag),
    IntParam(status_value),
    ..scope_params
  ]
  use totals <- result.try(database.query_on(
    conn,
    "SELECT COUNT(*) FROM inbound_orders WHERE " <> filter,
    params,
    {
      use total <- decode.field(0, json_codec.int())
      decode.success(total)
    },
  ))
  let total = case totals {
    [value, ..] -> value
    [] -> 0
  }
  use items <- result.try(database.query_on(
    conn,
    "SELECT bill_id, bill_code, supplier_name, contact_person, phone,
            warehouse_name, warehouse_id, bill_date, status, status_name,
            total_qty, total_money, remark
     FROM inbound_orders
     WHERE " <> filter <> "
     ORDER BY bill_date DESC, bill_id DESC
     LIMIT ? OFFSET ?",
    list.append(params, [IntParam(limit), IntParam(offset)]),
    inbound_summary_decoder(),
  ))
  Ok(InboundPage(items: items, total: total))
}

pub fn get_inbound(
  conn: Conn,
  bill_id: Int,
) -> Result(Option(PurchaseOrder), DbError) {
  use headers <- result.try(database.query_on(
    conn,
    "SELECT bill_id, bill_code, supplier_name, contact_person, phone,
            warehouse_name, warehouse_id, bill_date, status, status_name,
            total_qty, total_money, remark
     FROM inbound_orders WHERE bill_id = ?",
    [IntParam(bill_id)],
    inbound_summary_decoder(),
  ))
  case headers {
    [] -> Ok(None)
    [summary, ..] -> {
      use items <- result.try(database.query_on(
        conn,
        "SELECT bill_detail_id, sku_id, goods_code, goods_name, barcode, spec,
                package_qty, package_unit, base_qty, base_unit, price, total_money,
                position_code, area_code
         FROM inbound_order_items
         WHERE bill_id = ?
         ORDER BY line_no",
        [IntParam(bill_id)],
        inbound_item_decoder(),
      ))
      Ok(Some(PurchaseOrder(
        bill_id: summary.bill_id,
        bill_code: summary.bill_code,
        supplier_name: summary.supplier_name,
        contact_person: summary.contact_person,
        phone: summary.phone,
        warehouse_name: summary.warehouse_name,
        warehouse_id: summary.warehouse_id,
        bill_date: summary.bill_date,
        status: summary.status,
        status_name: summary.status_name,
        total_qty: summary.total_qty,
        total_money: summary.total_money,
        remark: summary.remark,
        items: items,
      )))
    }
  }
}

pub fn needs_inbound_detail(
  previous: Option(OrderStamp),
  order: PurchaseOrderSummary,
) -> Bool {
  case previous {
    None -> True
    Some(stamp) ->
      stamp.detail_synced_at == ""
      || stamp.status != order.status
      || stamp.bill_date != order.bill_date
      || !close_enough(stamp.total_qty, order.total_qty)
      || !close_enough(stamp.total_money, order.total_money)
  }
}

pub fn get_cursor(conn: Conn, entity: String) -> Result(String, DbError) {
  use rows <- result.try(database.query_on(
    conn,
    "SELECT cursor_value FROM sync_cursors WHERE entity = ?",
    [TextParam(entity)],
    {
      use value <- decode.field(0, decode.string)
      decode.success(value)
    },
  ))
  case rows {
    [] -> Ok("")
    [value, ..] -> Ok(value)
  }
}

pub fn set_cursor(
  conn: Conn,
  entity: String,
  cursor: String,
  updated_at: String,
) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO sync_cursors (entity, cursor_value, updated_at)
     VALUES (?, ?, ?)
     ON CONFLICT (entity) DO UPDATE SET
       cursor_value = EXCLUDED.cursor_value,
       updated_at = EXCLUDED.updated_at",
    [TextParam(entity), TextParam(cursor), TextParam(updated_at)],
  )
}

pub fn insert_run(conn: Conn, run: SyncRun) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO sync_runs (
       id, entity, trigger_kind, started_at, finished_at, status,
       fetched_count, upserted_count, error_message
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
    [
      TextParam(run.id),
      TextParam(run.entity),
      TextParam(run.trigger),
      TextParam(run.started_at),
      TextParam(run.finished_at),
      TextParam(run.status),
      IntParam(run.fetched),
      IntParam(run.upserted),
      TextParam(run.error_message),
    ],
  )
}

pub fn list_runs(conn: Conn) -> Result(List(SyncRun), DbError) {
  database.query_on(
    conn,
    "SELECT id, entity, trigger_kind, started_at, finished_at, status,
            fetched_count, upserted_count, error_message
     FROM sync_runs
     ORDER BY started_at DESC
     LIMIT 20",
    [],
    {
      use id <- decode.field(0, decode.string)
      use entity <- decode.field(1, decode.string)
      use trigger <- decode.field(2, decode.string)
      use started_at <- decode.field(3, decode.string)
      use finished_at <- decode.field(4, decode.string)
      use status <- decode.field(5, decode.string)
      use fetched <- decode.field(6, json_codec.int())
      use upserted <- decode.field(7, json_codec.int())
      use error_message <- decode.field(8, decode.string)
      decode.success(SyncRun(
        id: id,
        entity: entity,
        trigger: trigger,
        started_at: started_at,
        finished_at: finished_at,
        status: status,
        fetched: fetched,
        upserted: upserted,
        error_message: error_message,
      ))
    },
  )
}

pub fn needs_detail(previous: Option(OrderStamp), order: SalesOrderSummary) -> Bool {
  case previous {
    None -> True
    Some(stamp) ->
      stamp.detail_synced_at == ""
      || stamp.status != order.status
      || stamp.bill_date != order.bill_date
      || !close_enough(stamp.total_qty, order.total_qty)
      || !close_enough(stamp.total_money, order.total_money)
  }
}

fn close_enough(left: Float, right: Float) -> Bool {
  float.absolute_value(left -. right) <. 0.0001
}

fn stamp_decoder() -> decode.Decoder(OrderStamp) {
  use status <- decode.field(0, json_codec.int())
  use total_qty <- decode.field(1, json_codec.number())
  use total_money <- decode.field(2, json_codec.number())
  use bill_date <- decode.field(3, decode.string)
  use detail_synced_at <- decode.field(4, decode.string)
  decode.success(OrderStamp(
    status: status,
    total_qty: total_qty,
    total_money: total_money,
    bill_date: bill_date,
    detail_synced_at: detail_synced_at,
  ))
}

fn inbound_params(
  order: PurchaseOrderSummary,
  synced_at: String,
) -> List(database.Param) {
  [
    IntParam(order.bill_id),
    TextParam(order.bill_code),
    TextParam(order.supplier_name),
    TextParam(order.contact_person),
    TextParam(order.phone),
    TextParam(order.warehouse_name),
    IntParam(order.warehouse_id),
    TextParam(order.bill_date),
    IntParam(order.status),
    TextParam(order.status_name),
    FloatParam(order.total_qty),
    FloatParam(order.total_money),
    TextParam(order.remark),
    TextParam(synced_at),
  ]
}

fn insert_inbound_item(
  conn: Conn,
  bill_id: Int,
  line_no: Int,
  item: PurchaseOrderItem,
) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO inbound_order_items (
       bill_id, line_no, bill_detail_id, sku_id, goods_code, goods_name, barcode, spec,
       package_qty, package_unit, base_qty, base_unit, price, total_money,
       position_code, area_code
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
    [
      IntParam(bill_id),
      IntParam(line_no),
      IntParam(item.bill_detail_id),
      IntParam(item.sku_id),
      TextParam(item.goods_code),
      TextParam(item.goods_name),
      TextParam(item.barcode),
      TextParam(item.spec),
      FloatParam(item.package_qty),
      TextParam(item.package_unit),
      FloatParam(item.base_qty),
      TextParam(item.base_unit),
      FloatParam(item.price),
      FloatParam(item.total_money),
      TextParam(item.position_code),
      TextParam(item.area_code),
    ],
  )
}

fn inbound_summary_decoder() -> decode.Decoder(PurchaseOrderSummary) {
  use bill_id <- decode.field(0, json_codec.int())
  use bill_code <- decode.field(1, decode.string)
  use supplier_name <- decode.field(2, decode.string)
  use contact_person <- decode.field(3, decode.string)
  use phone <- decode.field(4, decode.string)
  use warehouse_name <- decode.field(5, decode.string)
  use warehouse_id <- decode.field(6, json_codec.int())
  use bill_date <- decode.field(7, decode.string)
  use status <- decode.field(8, json_codec.int())
  use status_name <- decode.field(9, decode.string)
  use total_qty <- decode.field(10, json_codec.number())
  use total_money <- decode.field(11, json_codec.number())
  use remark <- decode.field(12, decode.string)
  decode.success(PurchaseOrderSummary(
    bill_id: bill_id,
    bill_code: bill_code,
    supplier_name: supplier_name,
    contact_person: contact_person,
    phone: phone,
    warehouse_name: warehouse_name,
    warehouse_id: warehouse_id,
    bill_date: bill_date,
    status: status,
    status_name: status_name,
    total_qty: total_qty,
    total_money: total_money,
    remark: remark,
    ifred: 0,
  ))
}

fn inbound_item_decoder() -> decode.Decoder(PurchaseOrderItem) {
  use bill_detail_id <- decode.field(0, json_codec.int())
  use sku_id <- decode.field(1, json_codec.int())
  use goods_code <- decode.field(2, decode.string)
  use goods_name <- decode.field(3, decode.string)
  use barcode <- decode.field(4, decode.string)
  use spec <- decode.field(5, decode.string)
  use package_qty <- decode.field(6, json_codec.number())
  use package_unit <- decode.field(7, decode.string)
  use base_qty <- decode.field(8, json_codec.number())
  use base_unit <- decode.field(9, decode.string)
  use price <- decode.field(10, json_codec.number())
  use total_money <- decode.field(11, json_codec.number())
  use position_code <- decode.field(12, decode.string)
  use area_code <- decode.field(13, decode.string)
  decode.success(PurchaseOrderItem(
    bill_detail_id: bill_detail_id,
    sku_id: sku_id,
    goods_code: goods_code,
    goods_name: goods_name,
    barcode: barcode,
    spec: spec,
    package_qty: package_qty,
    package_unit: package_unit,
    base_qty: base_qty,
    base_unit: base_unit,
    price: price,
    total_money: total_money,
    position_code: position_code,
    area_code: area_code,
  ))
}

fn order_params(order: SalesOrderSummary, synced_at: String) -> List(database.Param) {
  [
    IntParam(order.bill_id),
    TextParam(order.bill_code),
    TextParam(order.customer_name),
    TextParam(order.contact_person),
    TextParam(order.phone),
    TextParam(order.warehouse_name),
    IntParam(order.warehouse_id),
    TextParam(order.bill_date),
    IntParam(order.status),
    TextParam(order.status_name),
    FloatParam(order.total_qty),
    FloatParam(order.total_money),
    TextParam(order.remark),
    TextParam(synced_at),
  ]
}

fn insert_item(
  conn: Conn,
  bill_id: Int,
  line_no: Int,
  item: SalesOrderItem,
) -> Result(Nil, DbError) {
  database.exec_on(
    conn,
    "INSERT INTO outbound_order_items (
       bill_id, line_no, bill_detail_id, sku_id, goods_code, goods_name, barcode, spec,
       package_qty, package_unit, base_qty, base_unit, price, total_money,
       position_code, area_code
     ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
    [
      IntParam(bill_id),
      IntParam(line_no),
      IntParam(item.bill_detail_id),
      IntParam(item.sku_id),
      TextParam(item.goods_code),
      TextParam(item.goods_name),
      TextParam(item.barcode),
      TextParam(item.spec),
      FloatParam(item.package_qty),
      TextParam(item.package_unit),
      FloatParam(item.base_qty),
      TextParam(item.base_unit),
      FloatParam(item.price),
      FloatParam(item.total_money),
      TextParam(item.position_code),
      TextParam(item.area_code),
    ],
  )
}

fn product_decoder() -> decode.Decoder(GoodsItem) {
  use id <- decode.field(0, json_codec.int())
  use code <- decode.field(1, decode.string)
  use name <- decode.field(2, decode.string)
  use barcode <- decode.field(3, decode.string)
  use spec <- decode.field(4, decode.string)
  use unit <- decode.field(5, decode.string)
  use shelf_life <- decode.field(6, decode.string)
  use shelf_life_days <- decode.field(7, json_codec.int())
  use position_code <- decode.field(8, decode.string)
  use disabled <- decode.field(9, json_codec.int())
  use is_category <- decode.field(10, json_codec.int())
  use modified_at <- decode.field(11, decode.string)
  use created_at <- decode.field(12, decode.string)
  decode.success(GoodsItem(
    id: id,
    code: code,
    name: name,
    barcode: barcode,
    unit: unit,
    spec: spec,
    shelf_life: shelf_life,
    shelf_life_days: shelf_life_days,
    position_code: position_code,
    disabled: disabled != 0,
    is_category: is_category != 0,
    modified_at: modified_at,
    created_at: created_at,
  ))
}

fn summary_decoder() -> decode.Decoder(SalesOrderSummary) {
  use bill_id <- decode.field(0, json_codec.int())
  use bill_code <- decode.field(1, decode.string)
  use customer_name <- decode.field(2, decode.string)
  use contact_person <- decode.field(3, decode.string)
  use phone <- decode.field(4, decode.string)
  use warehouse_name <- decode.field(5, decode.string)
  use warehouse_id <- decode.field(6, json_codec.int())
  use bill_date <- decode.field(7, decode.string)
  use status <- decode.field(8, json_codec.int())
  use status_name <- decode.field(9, decode.string)
  use total_qty <- decode.field(10, json_codec.number())
  use total_money <- decode.field(11, json_codec.number())
  use remark <- decode.field(12, decode.string)
  decode.success(SalesOrderSummary(
    bill_id: bill_id,
    bill_code: bill_code,
    customer_name: customer_name,
    contact_person: contact_person,
    phone: phone,
    warehouse_name: warehouse_name,
    warehouse_id: warehouse_id,
    bill_date: bill_date,
    status: status,
    status_name: status_name,
    total_qty: total_qty,
    total_money: total_money,
    remark: remark,
  ))
}

fn item_decoder() -> decode.Decoder(SalesOrderItem) {
  use bill_detail_id <- decode.field(0, json_codec.int())
  use sku_id <- decode.field(1, json_codec.int())
  use goods_code <- decode.field(2, decode.string)
  use goods_name <- decode.field(3, decode.string)
  use barcode <- decode.field(4, decode.string)
  use spec <- decode.field(5, decode.string)
  use package_qty <- decode.field(6, json_codec.number())
  use package_unit <- decode.field(7, decode.string)
  use base_qty <- decode.field(8, json_codec.number())
  use base_unit <- decode.field(9, decode.string)
  use price <- decode.field(10, json_codec.number())
  use total_money <- decode.field(11, json_codec.number())
  use position_code <- decode.field(12, decode.string)
  use area_code <- decode.field(13, decode.string)
  decode.success(SalesOrderItem(
    bill_detail_id: bill_detail_id,
    sku_id: sku_id,
    goods_code: goods_code,
    goods_name: goods_name,
    barcode: barcode,
    spec: spec,
    package_qty: package_qty,
    package_unit: package_unit,
    base_qty: base_qty,
    base_unit: base_unit,
    price: price,
    total_money: total_money,
    position_code: position_code,
    area_code: area_code,
  ))
}

fn like_pattern(keyword: String) -> String {
  let cleaned =
    keyword
    |> string.replace("%", "")
    |> string.replace("_", "")
  "%" <> cleaned <> "%"
}

fn bool_int(value: Bool) -> Int {
  case value {
    True -> 1
    False -> 0
  }
}