logos-delivery/logos_delivery/channels/reliable_channel_manager.nim

## Reliable Channel API entry point.
##
## Owns the set of `ReliableChannel` instances and exposes lifecycle and
## send/receive operations addressed by `ChannelId`.
##
## See: https://lip.logos.co/messaging/raw/reliable-channel-api.html

import std/[options, tables]
import results
import chronos
import chronicles
import stew/byteutils

import brokers/broker_implement

import logos_delivery/api/messaging_client_interface
import logos_delivery/api/reliable_channel_manager_interface
import logos_delivery/waku/events/message_events as waku_message_events
import logos_delivery/waku/waku_core/topics
import logos_delivery/waku/persistency/sds_persistency

import ./reliable_channel
import ./encryption/noop_encryption

export reliable_channel

const SdsJobId = "sds"
  ## One persistency job shared by every channel's SDS state; rows are
  ## keyed by channelId.

type ReliableChannelManager* = ref object of ReliableChannelManagerInterface
  channels: Table[ChannelId, ReliableChannel]
  messagingClient: MessagingClientInterface
    ## Borrowed from the owning `Waku`.
    ## Default egress dispatch for channels created through this manager.
    ## Constructed at mount time as a closure over `MessagingClient.send`
    ## so the channel layer itself stays callable-only.

proc start*(self: ReliableChannelManager): Result[void, string] =
  ## Placeholder: per-channel listeners are installed in `ReliableChannel.new`,
  ## so the manager has nothing to start at this layer. Kept for symmetry
  ## with the `Waku` mount/start lifecycle and as a hook for future state.
  discard
  ok()

proc stop*(self: ReliableChannelManager) {.async.} =
  ## Stops every channel's SDS background loops. Persisted state survives.
  for chn in self.channels.values:
    await chn.stop()
  self.channels.clear()

proc sdsPersistence(): Option[Persistence] =
  ## SDS backend from the Persistency singleton; memory-only fallback when
  ## it is unavailable (e.g. unit tests).
  let p = Persistency.instance().valueOr:
    info "SDS persistence disabled, running memory-only", reason = $error
    return none(Persistence)
  let job = p.openJob(SdsJobId).valueOr:
    warn "SDS persistence disabled, could not open persistency job",
      jobId = SdsJobId, reason = $error
    return none(Persistence)
  return some(newSdsPersistence(job))

BrokerImplement ReliableChannelManager of ReliableChannelManagerInterface:
  proc new(
      T: typedesc[ReliableChannelManager], messagingClient: MessagingClientInterface
  ): T =
    T(
      channels: initTable[ChannelId, ReliableChannel](),
      messagingClient: messagingClient,
    )

  method createReliableChannel*(
      self: ReliableChannelManager,
      channelId: ChannelId,
      contentTopic: ContentTopic,
      senderId: SdsParticipantID,
  ): Future[Result[ChannelId, string]] {.async.} =
    ## Spec entry point. The`rng` the channel needs are
    ## sourced from the owning `ReliableChannelManager` rather than passed
    ## per call. Encryption is wired up through the `Encrypt`/`Decrypt`
    ## request brokers — the application installs its own providers
    ## (or `setNoopEncryption()`) before traffic flows.
    ##
    ## Segmentation, SDS and rate-limit configs will eventually be read
    ## from the node's `NodeConfig`. Defaults for now.
    if self.channels.hasKey(channelId):
      return err("channel already exists: " & channelId)

    let segConfig = SegmentationConfig(
      segmentSizeBytes: DefaultSegmentSizeBytes,
      enableReedSolomon: false,
      persistence: nil,
    )
    let sdsConfig = SdsConfig(
      acknowledgementTimeoutMs: DefaultAcknowledgementTimeoutMs,
      maxRetransmissions: DefaultMaxRetransmissions,
      causalHistorySize: DefaultCausalHistorySize,
      persistence: sdsPersistence(),
    )
    let rateConfig = RateLimitConfig(
      epochPeriodSec: DefaultEpochPeriodSec, messagesPerEpoch: DefaultMessagesPerEpoch
    )

    let chn = ReliableChannel.new(
      messagingClient = self.messagingClient,
      channelId = channelId,
      contentTopic = contentTopic,
      senderId = senderId,
      segConfig = segConfig,
      sdsConfig = sdsConfig,
      rateConfig = rateConfig,
      brokerCtx = self.brokerCtx,
    )

    self.channels[channelId] = chn
    return ok(channelId)

  method closeChannel*(
      self: ReliableChannelManager, channelId: ChannelId
  ): Future[Result[void, string]] {.async.} =
    ## Stops the channel's SDS loops and releases the channel. Persisted SDS
    ## state survives, so re-creating the channel restores it.
    let chn = self.channels.getOrDefault(channelId)
    if chn.isNil():
      return err("unknown channel: " & channelId)
    self.channels.del(channelId)
    await chn.stop()
    return ok()

  method sendOnChannel(
      self: ReliableChannelManager,
      channelId: ChannelId,
      appPayload: seq[byte],
      ephemeral: bool = false,
  ): Future[Result[RequestId, string]] {.async.} =
    ## Spec-level entry point. Looks the channel up by id and delegates
    ## to `ReliableChannel.send`, which exposes the visible pipeline
    ## segmentation -> sds -> rate_limit_manager -> encryption.
    let chn = self.channels.getOrDefault(channelId)
    if chn.isNil():
      return err("unknown channel: " & channelId)
    return await chn.send(appPayload, ephemeral)

## Inbound messages are not handed to the manager by direct call. Each
## `ReliableChannel` installs its own `MessageReceivedEvent` listener
## in `ReliableChannel.new`, filters by spec marker and `contentTopic`,
## and routes to its private `onMessageReceived`. This keeps the lower
## layer (MessagingClient/Waku) unaware of the existence of ReliableChannel
## and keeps the manager out of per-channel event dispatch.