src/orders/orders.oxl

use super.model.*
use super.limits { ORDER_SHARDS, PAGE_SIZE, ORDER_EVENTS_RETENTION, STATS_BUFFER }
use std.orm { Db }
use std.stream { Broadcast, filter }
use std.time { today, Tz, ZonedTime, AmbiguousTime }

// @owner: пользователь видит только свои заказы; @shard: таблица разбита по userId.
@table("orders") @owner(.userId, bypass: [.Support, .Admin])
@index(.userId, .createdAt) @shard(by: .userId, count: ORDER_SHARDS)
pub(api) data Order {
  @pk pub(api) id: OrderId
  pub(api) userId: UserId
  pub(api) items: List<Item>          // хранится сериализованным в одной колонке
  pub(api) total: Money
  pub(api) receiptId: str             // id платежа во внешней системе
  pub(api) status: Status = Status.New
  pub(api) createdAt: Time = now()
}

@derive(Clone) pub(api) enum OrderEvent {
  Created(id: OrderId, userId: UserId)
  StatusChanged(id: OrderId, userId: UserId, status: Status)
}

// Топик: каждое событие получают все подписчики; порядок гарантирован внутри userId
@retain(ORDER_EVENTS_RETENTION)
pub(api) topic OrderEvents: OrderEvent

@derive(Show) pub(api) data Stats { pub(api) ordersToday: u64; pub(api) revenueToday: Money }

// Топик статистики хранит только последнее значение
@retain(count: 1) pub(api) topic StatsUpdates: Stats

@v(1) pub(api) service Orders {
  @readonly rpc recent(page: u32) -> List<Order>
  @readonly rpc get(id: OrderId) -> Order | NotFound
  // internal: вызывается только из checkout, внешним клиентам недоступен
  @idempotent @internal rpc create(snap: CartSnapshot, receiptId: str) -> Order
  // resumable: при обрыве клиент переподключится и продолжит с последнего offset
  @resumable rpc watch() -> stream<OrderEvent, TopicError>
  rpc liveStats() -> stream<Shared<Stats>, Lagged>
  // ОТКЛОНЕНИЕ (DEVIATIONS.md #7): в Стандарте статистику публикует `main` (leader + топик StatsUpdates); без топиков и
  // `dep` уровня модуля её считает внутренний rpc, который вызывает `main`.
  @internal rpc publishStats() -> ()
}

impl Orders {
  dep db: Db

  rpc recent(page) {
    let uid = callerId(ctx)?          // AccessDenied, если вызывает не пользователь
    from o in db.Order where o.userId == uid
      order by o.createdAt desc skip page * PAGE_SIZE take PAGE_SIZE
  }

  rpc get(id) { db.find(Order, id) ?? fail NotFound{ what: "order", id: id.raw } }

  rpc create(snap, receiptId) {
    let o = Order{ id: OrderId(cluster.ids.next() as u64), userId: snap.userId,
      items: snap.items, total: snap.total, receiptId }
    // заказ и событие — одна транзакция: событие пишется в outbox и публикуется после коммита (at-least-once). Коммит, ответ на
    // который потерян при смене лидера шарда, сверяется с новым лидером по id транзакции (D-5521).
    atomic(db) {
      db.insert(o)
      emit OrderEvents(OrderEvent.Created(o.id, o.userId))
    }
    o
  }

  // Подписка только на события своего пользователя, начиная с новых
  rpc watch() {
    let uid = match callerId(ctx) {
      UserId u => u
      AccessDenied _ => UserId(0)
    }
    filter(OrderEvents.subscribe(from: TopicStart.Latest), e => ownerOf(e) == uid)
  }

  rpc liveStats() { LIVE_STATS.subscribe() }

  rpc publishStats() { _ = StatsUpdates.publish(computeStats(db)) }
}

// Локальный broadcast-хаб на узле: медленный клиент теряет старые значения (DropOldest), новый сразу получает последнее.
pub static LIVE_STATS: Broadcast<Stats> = Broadcast<Stats>(buffer: STATS_BUFFER, slow: .DropOldest, replay: 1)

policy Orders {
  allow recent, get, watch if caller is .User(..)
  allow liveStats if caller.has(role: .Support) || caller.has(role: .Admin)
}

fn ownerOf(e: &OrderEvent) -> UserId {
  match e {
    OrderEvent.Created(id: _, userId: u) => u
    OrderEvent.StatusChanged(id: _, userId: u, status: _) => u
  }
}

// Подписка на топик статистики → локальный broadcast (в V.5 это тело задачи в `main`; импорт топика по имени работает с D-5491, перенос в `main` не делался)
pub fn relayStats() -> () | TopicError  db, net, time {
  for await s in StatsUpdates.subscribe(from: TopicStart.Latest) { LIVE_STATS.publish(s) }
}

// Обычная функция модуля; db передаётся явно заимствованием
pub fn computeStats(db: &Db) -> Stats  io, db, time {
  // начало сегодняшнего дня по часовому поясу узла
  let since = match today().atStartOfDay().in(Tz.node()) {
    ZonedTime z => z.time
    AmbiguousTime _ => now() - 24h
  }
  let todays = from o in db.Order where o.createdAt >= since
  Stats{ ordersToday: todays.len() as u64,
    revenueToday: todays.fold(Money.ZERO, (a, o) => a + o.total) }
}