syndicate-nim/src/syndicate/actors.nim

# SPDX-FileCopyrightText: ☭ Emery Hemingway
# SPDX-License-Identifier: Unlicense

import std/[deques, hashes, options, times]
import pkg/cps
import preserves
import ../syndicate/protocols/[protocol, sturdy]

# const traceSyndicate {.booldefine.}: bool = true
const traceSyndicate* = true

when traceSyndicate:
  import std/streams
  from std/os import getEnv
  import ./protocols/trace

export protocol.Handle

type
  Actor* = ref object
    ## https://synit.org/book/glossary.html#actor
    root: Facet
    handleAllocator: Handle
    id: ActorId
    when traceSyndicate:
      traceStream: FileStream

  Facet* = ref object
    ## https://synit.org/book/glossary.html#facet
    actor: Actor
    parent: Facet
    stopHandlers: Work
    state: FacetState
    when traceSyndicate:
      id: FacetId

  FacetState = enum fIdle, fRunning, fStopped

  Entity* = ref object of RootObj
    ## https://synit.org/book/glossary.html#entity
    facet*: Facet
    oid*: sturdy.Oid # oid is how Entities are identified over the wire
    when traceSyndicate:
      id: FacetId

  Cap* {.final, preservesEmbedded.} = ref object of EmbeddedObj
    relay*: Facet
    target*: Entity
    attenuation*: seq[sturdy.Caveat]

  Turn* = ref object
    ## https://synit.org/book/glossary.html#turn
    facet: Facet
    entity: Entity
    event: Option[protocol.Event]
    work: Work
    when traceSyndicate:
      desc: TurnDescription

  Work = Deque[Cont]

  Cont* = ref object of Continuation
    facet: Facet
    turn: Turn

using
  actor: Actor
  facet: Facet
  entity: Entity
  cap: Cap
  turn: Turn

proc hash*(facet): Hash = facet.unsafeAddr.hash
proc hash*(cap): Hash = cap.unsafeAddr.hash

proc newFacet(actor: Actor; parent: Facet): Facet =
  result = Facet(
      actor: actor,
      parent: parent,
    )
  when traceSyndicate:
    if not parent.isNil:
      result.id = parent.id
      inc(result.id.register)

proc newActor(name: string): Actor =
  result = Actor(id: name.toPreserves)
  result.root = newFacet(result, nil)
  when traceSyndicate:
    let path = getEnv("SYNDICATE_TRACE_FILE", "")
    case path
    of "": discard
    of "-": result.traceStream = newFileStream(stderr)
    else: result.traceStream = openFileStream(path, fmWrite)

when traceSyndicate:
  proc trace(actor: Actor; act: ActorActivation) =
    if not actor.traceStream.isNil:
      var entry = TraceEntry(
          timestamp: getTime().toUnixFloat(),
          actor: actor.id,
          item: act,
        )
      actor.traceStream.writeLine($entry.toPreserves)

  proc traceTarget(cap): trace.Target =
    Target(
        actor: cap.relay.actor.id,
        facet: cap.relay.id,
        oid: cap.target.oid.toPreserves,
      )

proc startActor(actor: Actor) =
  when traceSyndicate:
    var act = ActorActivation(orkind: ActorActivationKind.start)
    trace(actor, act)

proc stopActor(actor: Actor) =
  when traceSyndicate:
    var act = ActorActivation(orkind: ActorActivationKind.stop)
    trace(actor, act)

proc bootActor*(name: string, c: Cont) =
  let actor = newActor(name)
  startActor(actor)

proc queue*(t: Turn; c: Cont) =
  assert not c.facet.isNil
  t.work.addLast(c)

template syndicate*(prc: typed): untyped =
  cps(Cont, prc)

proc activeFacet*(c: Cont): Facet {.cpsVoodoo.} =
  ## Return the active `Facet` within a `{.syndicate.}` context.
  assert not c.facet.isNil
  c.facet

proc activeTurn*(c: Cont): Turn {.cpsVoodoo.} =
  ## Return the active `Turn` within a `{.syndicate.}` context.
  assert not c.turn.isNil
  c.turn

type
  AssertionRef* = ref object
    value*: Value
      # if the Enity methods take a Value object then the generated
      # C code has "redefinition of struct" problems when orc is enabled

method publish*(e: Entity; turn: Turn; v: AssertionRef; h: Handle) {.base.} = discard
method retract*(e: Entity; turn: Turn; h: Handle) {.base.} = discard
method message*(e: Entity; turn: Turn; v: AssertionRef) {.base.} = discard
method sync*(e: Entity; turn: Turn; peer: Cap) {.base.} = discard

proc newCap*(f: Facet; e: Entity): Cap =
  Cap(relay: f, target: e)

proc nextHandle(facet: Facet): Handle =
  inc(facet.actor.handleAllocator)
  facet.actor.handleAllocator

proc publish*(turn: Turn; cap: Cap; val: Value): Handle =
  result = turn.facet.nextHandle()
  when traceSyndicate:
    var act = ActionDescription(orKind: ActionDescriptionKind.enqueue)
    act.enqueue.event = TargetedTurnEvent(
        target: cap.traceTarget,
        detail: trace.TurnEvent(orKind: trace.TurnEventKind.assert)
      )
    act.enqueue.event.detail = trace.TurnEvent(orKind: TurnEventKind.assert)
    act.enqueue.event.detail.assert = TurnEventAssert(
        assertion: AssertionDescription(orKind: AssertionDescriptionKind.value),
        handle: result,
      )
    act.enqueue.event.detail.assert.assertion.value.value = val
    turn.desc.actions.add act

proc retract*(t: Turn; h: Handle) =
  discard

proc message*(t: Turn; cap: Cap; val: Value) =
  discard

proc sync*(t: Turn) =
  discard

proc publish*(cap: Cap; val: Value): Handle {.syndicate.} =
  publish(activeTurn(), cap, val)

proc retract*(h: Handle) {.syndicate.} =
  activeTurn().retract(h)

proc message*(cap: Cap; val: Value) {.syndicate.} =
  activeTurn().message(cap, val)

proc sync*() {.syndicate.} =
  activeTurn().sync()