FIN 5.3.1 中文文档

外观

notificationExt 扩展参考 ​

原文: https://finproducts.atlassian.net/wiki/spaces/JIDD3/pages/34793685250/FIN+5.3.1+notificationExt 中文整理基于 FIN 5.3.1

导读： 抽象常量类 AbstractMailerTarget : NotificationTarget

以下为 API 参考，保留英文以便对照源码；目录已列出（英文标题便于检索）；概述已尽量中译。

目录（导读） ​

  • classes
  • Slot Details
    • _templateCache
    • _templateMime
    • generateBody
    • getAttachments
    • make
    • sendEmail
    • subject
    • templateCache
    • templateRef
    • body
    • mimeType
    • TYPE
    • execute
    • to
    • wrap
    • emailPart
    • files
    • toFinal
    • _topicCache
    • topic
    • shouldSilenceEmail
    • silenceEndTime
    • silenceStartTime
    • user
    • notifyTarget
    • TARGET_TAG_NAME
    • notify
    • notifyManual
    • onHandleNotification
    • onStart
    • onStop
    • web
    • addInterests
    • context
    • currentTries
    • incrementTries
    • interest
    • target
    • _sockets
    • lib
    • onGet
    • sendNotification
    • disabled
    • include
    • log
    • notificationOn
    • notificationTarget
    • onFilter
    • addDigest
    • addImmediate
    • digestUsers
    • immediateUsers
    • on
    • topicOn
    • CACHE_PERIOD
    • getDigestUsersForInterest
    • getImmediateUsersForInterest
    • isExpired
    • refreshIfExpired
    • topicName

概述（中译） ​

src 抽象常量类 AbstractMailerTarget : NotificationTarget

构造函数

字段

方法

插槽详情

src 受保护的 const AtomicRef _templateCache := AtomicRef.<ctor (Unsafe.<ctor ([:]))

src 受保护的常量 AtomicRef _templateMime := AtomicRef.<ctor ()

src 受保护的 ContentBody generateBody(HxContext context, Obj interest)

API 正文（英文） ​

getAttachments ​

src protected File[] getAttachments(HxContext context, Obj interest)

make ​

src new make(Dict rec, |This| f)

sendEmail ​

src protected Void sendEmail(HxContext context, Obj interest, Str[] to)

subject ​

src @TypedTag const Str subject

templateCache ​

src protected Str:TemplateProcessor templateCache()

templateRef ​

src @TypedTag const Ref? templateRef

ContentBody ​

src const class ContentBody : Obj

Constructors

make

Fields

body
mimeType

Slot Details ​

body ​

src const Str body

make ​

src new make(Str body, MimeType mimeType)

mimeType ​

src const MimeType mimeType

DirectEmailNotificationTarget ​

src const class DirectEmailNotificationTarget : AbstractMailerTarget

Constructors

make
wrap

Fields

TYPE
to

Methods

execute

Slot Details ​

TYPE ​

src const static Str TYPE := "directEmail"

execute ​

src virtual override Future execute(HxContext context, Obj interest)

make ​

src new make(Dict rec, |This| f)

to ​

src @TypedTag const Str to

wrap ​

src static new wrap(Dict? dict)

EmailParts ​

src class EmailParts : Obj

Constructors

make

Fields

emailPart
files

Methods

toFinal

Slot Details ​

emailPart ​

src TextPart emailPart

files ​

src FilePart[] files

make ​

src new make(ContentBody email, File[] attachments)

toFinal ​

src EmailPart toFinal()

EmailTopicNotificationTarget ​

src const class EmailTopicNotificationTarget : AbstractMailerTarget

Constructors

make
wrap

Fields

TYPE
_topicCache
topic

Methods

execute

Slot Details ​

TYPE ​

src const static Str TYPE := "emailTopic"

_topicCache ​

src const AtomicRef _topicCache := AtomicRef.<ctor>()

execute ​

src virtual override Future execute(HxContext context, Obj interest)

make ​

src new make(Dict rec, |This| f)

topic ​

src @TypedTag const Str topic

wrap ​

src static new wrap(Dict? dict)

EmailUser ​

src const class EmailUser : Obj

Constructors

make

Fields

silenceEndTime
silenceStartTime
user

Methods

shouldSilenceEmail

Slot Details ​

make ​

src new make(HxUser user)

shouldSilenceEmail ​

src Bool shouldSilenceEmail(Time when)

silenceEndTime ​

src const Time? silenceEndTime

silenceStartTime ​

src const Time? silenceStartTime

user ​

src const HxUser user

NotificationFuncs ​

src class NotificationFuncs : Obj

Methods

notifyTarget

Slot Details ​

notifyTarget ​

src @Axon static Void notifyTarget(Obj target, Obj interest)

NotificationLib ​

src const class NotificationLib : HxLib

Constructors

make

Fields

TARGET_TAG_NAME
web

Methods

notify
notifyManual
onHandleNotification
onStart
onStop

Slot Details ​

TARGET_TAG_NAME ​

src const static Str TARGET_TAG_NAME := "notificationTarget"

make ​

src new make()

notify ​

src Future notify(HxContext context, Str subject, Obj interest)

notifyManual ​

src Future notifyManual(HxContext context, Ref targetRef, Obj interest)

onHandleNotification ​

src protected Obj? onHandleNotification(NotificationQueueItem item)

onStart ​

src virtual override Void onStart()

onStop ​

src virtual override Void onStop()

web ​

src virtual const override NotificationRestAPI web := NotificationRestAPI.<ctor>(this)

NotificationQueueItem ​

src const class NotificationQueueItem : Obj

Constructors

make

Fields

target

Methods

addInterests
context
currentTries
incrementTries
interest

Slot Details ​

addInterests ​

src Void addInterests(Obj[] interests)

context ​

src HxContext context()

currentTries ​

src Int currentTries()

incrementTries ​

src Void incrementTries()

interest ​

src Obj interest()

make ​

src new make(HxContext context, Obj[] interest, NotificationTarget target)

target ​

src const NotificationTarget target

NotificationRestAPI ​

src const class NotificationRestAPI : HxLibWeb

Constructors

make

Fields

_sockets

Methods

lib
onGet
sendNotification

Slot Details ​

_sockets ​

src const Unsafe _sockets := Unsafe.<ctor>([,])

lib ​

src virtual override NotificationLib lib()

make ​

src new make(NotificationLib lib)

onGet ​

src virtual override Void onGet()

sendNotification ​

src Void sendNotification(Str subject, Obj notification)

NotificationTarget ​

src abstract const class NotificationTarget : TypedDict

Constructors

make

Fields

disabled
log
notificationOn
notificationTarget
onFilter

Methods

execute
include

Slot Details ​

disabled ​

src @TypedTag const Bool disabled

execute ​

src abstract Future execute(HxContext context, Obj interest)

include ​

src Bool include(HxContext context, Obj interest)

log ​

src protected const Log log := ...

make ​

src new make(Dict rec, |This| f)

notificationOn ​

src @TypedTag const Str? notificationOn

notificationTarget ​

src @TypedTag const Str notificationTarget

onFilter ​

src const Filter? onFilter

PushBulletNotificationTarget ​

src const class PushBulletNotificationTarget : NotificationTarget

Constructors

make
wrap

Fields

TYPE

Methods

execute

Slot Details ​

TYPE ​

src const static Str TYPE := "pushBullet"

execute ​

src virtual override Future execute(HxContext context, Obj interest)

make ​

src new make(Dict rec, |This| f)

wrap ​

src static new wrap(Dict? dict)

Topic ​

src const class Topic : TypedDict

Constructors

make
wrap

Fields

on
topicOn

Methods

addDigest
addImmediate
digestUsers
immediateUsers

Slot Details ​

addDigest ​

src Void addDigest(EmailUser user)

addImmediate ​

src Void addImmediate(EmailUser user)

digestUsers ​

src EmailUser[]? digestUsers()

immediateUsers ​

src EmailUser[]? immediateUsers()

make ​

src new make(Dict rec, |This| f)

on ​

src const Filter on

topicOn ​

src @TypedTag const Str topicOn

wrap ​

src static new wrap(Dict dict)

TopicCache ​

src class TopicCache : Obj

Constructors

make

Fields

CACHE_PERIOD
topicName

Methods

getDigestUsersForInterest
getImmediateUsersForInterest
isExpired
refreshIfExpired

Slot Details ​

CACHE_PERIOD ​

src const static Duration CACHE_PERIOD := 27sec

getDigestUsersForInterest ​

src HxUser[] getDigestUsersForInterest(HxContext context, Obj interest)

getImmediateUsersForInterest ​

src HxUser[] getImmediateUsersForInterest(HxContext context, Obj interest)

isExpired ​

src Bool isExpired()

make ​

src new make(HxContext context, Str topicName)

refreshIfExpired ​

src Void refreshIfExpired()

topicName ​

src const Str topicName

using haystack
using concurrent
using hx
using hxIO
using skyarcd
using email
using inet
using mustacheExt
using finSmtp

const abstract class AbstractMailerTarget : NotificationTarget 
{
  @TypedTag
  const Str subject
  
  @TypedTag
  const Ref? templateRef
  
  protected const AtomicRef _templateCache := AtomicRef(Unsafe([:]))
  protected const AtomicRef _templateMime := AtomicRef()
  
  new make(Dict rec, |This| f): super(rec, f) {}
  
  protected ContentBody generateBody(HxContext context, Obj interest)
  {
    // Process template
    itemType := interest.typeof.name
    
    if (interest is List && ((List) interest).size == 1)
      interest = ((List) interest).get(0)
            
    if (!this.templateCache.containsKey(itemType))
    {
      // Get user defined template
      Str? template
      Str:Str partials := [:]
      
      IOHandle? templateHandle
      
      if (this.templateRef != null)
      {
        templateRec := context.db.readById(this.templateRef)
        templateHandle = IOHandle.fromObj(context.rt, templateRec)
        
        binTag := templateRec.get("file") as Bin
        this._templateMime.getAndSet(binTag.mime)
      }
      else // Get Built in Templates
      if (interest is Grid || interest is List)
      {
        if (interest is Grid && ((Grid) interest).meta.has("digest"))
        {
          listPartial := IOHandle.fromObj(context.rt, `fan://notificationExt/resources/alarming/DefaultEmailListPartial.html`)
          templateHandle = IOHandle.fromObj(context.rt, `fan://notificationExt/resources/alarming/DefaultEmailDigestTemplate.html`)
          
          partials.set("DefaultEmailListPartial", listPartial.inToBuf.readAllStr)
        }
        else
          templateHandle = IOHandle.fromObj(context.rt, `fan://notificationExt/resources/alarming/DefaultEmailListTemplate.html`)
        
        this._templateMime.getAndSet(MimeType.fromStr("text/html"))
      }
      else
      {
        alarmPartial := IOHandle.fromObj(context.rt, `fan://notificationExt/resources/alarming/DefaultAlarmPartial.html`)
        templateHandle = IOHandle.fromObj(context.rt, `fan://notificationExt/resources/alarming/DefaultEmailTemplate.html`)
        
        partials.set("DefaultAlarmPartial", alarmPartial.inToBuf.readAllStr)
        
        this._templateMime.getAndSet(MimeType.fromStr("text/html"))
      }
      
      template = templateHandle.inToBuf.readAllStr
            
      processor := TemplateProcessor(context)
      processor.parseTemplate(template, partials)
      
      templateCache.set(itemType, processor)
    }
    
    // Remove "info" key from metadata, we don’t need the data for every variable in the program.”
    if (interest is Dict && ((Dict)interest).has("metadata"))
    { 
      metadata := ((Dict)interest).get("metadata")
      if (metadata is Dict)
      {
        filtered := Etc.dictRemove(metadata, "info")
        interest = Etc.dictSet(interest, "metadata", filtered)
      }
    }
  
    return ContentBody(
      this.templateCache.get(itemType).applyTemplate(interest),
      this._templateMime.val
    )
  }
  
  protected File[] getAttachments(HxContext context, Obj interest)
  {
    File[] files := [,]
    
    if (this.templateRef != null)
    {
      // Template
      templateRec := context.db.readById(this.templateRef)
      ((Context) context).asCur | -> Void |
      {
        templateRec.each | value, tagName | {
          if (tagName == "file")
            return
          
          if (value is Bin)
          {
            handle := (IOHandle) IOFuncs.ioBin(templateRec, tagName)
            
            files.add(
              handle.inToBuf.toFile(
                "rec://${templateRef.id}/${tagName}".toUri
              )
            )
          }
        }
      }
    }
    else
    {
      if (interest is Grid)
      {
        if (((Grid) interest).meta.has("digest"))
        {
          interest = ((Grid) interest).toRows.flatMap |row| { row.get("alarms")  }
        }
        else
          interest = ((Grid) interest).toRows
      }
      
      if (interest is Dict)
        interest = [interest]
      
      Dict[] items := interest
      
      Str:Str? stateNames := [:]
      Str:Str? kindNames := [:]
      
      items.each | item | {
        
        // Set State
        state := ((Str?) item.get("state")) ?: "unknown"
  
        if (item.has("acked"))
          state = "acknowledged${state}"
        
        stateNames.set(state, null)
        
        // Set Kinds (From Alarm Metadata [tags])
        if (item.has("metadata"))
        {
          Obj metadata := item.get("metadata")
          
          if (metadata is Dict)
          {
            ((Dict) metadata).each | value | {
              kind := Kind.fromVal(value)
              kindNames.set(kind.name, null)
            }
          }
        }
      }
      
      // Get State Icons
      stateNames.keys.each | state | {
        try
        {
          stateIcon := IOHandle.fromObj(
            context.rt,
            "fan://notificationExt/resources/alarming/${state}Icon.png".toUri
          )
          
          files.add(
            stateIcon.inToBuf.toFile(
              "fan://notificationExt/resources/alarming/${state}Icon.png".toUri
            )
          )
        }
        catch (Err error)
        {
          // If failed to find state icon, default to off normal alarm icon
          
          stateIcon := IOHandle.fromObj(
            context.rt,
            "fan://notificationExt/resources/alarming/offNormalIcon.png".toUri
          )
          
          files.add(
            stateIcon.inToBuf.toFile(
              "fan://notificationExt/resources/alarming/${state}Icon.png".toUri
            )
          )
        }
      }

      // Get Kind Icons
      kindNames.keys.each | kindName | {
        try
        {
          typeHandle := IOHandle.fromObj(
            context.rt,
            "fan://notificationExt/resources/alarming/${kindName}Icon.png".toUri
          )
          
          files.add(
            typeHandle.inToBuf.toFile(
              "fan://notificationExt/resources/alarming/${kindName}Icon.png".toUri
            )
          )
        }
        catch (Err error)
        {
          // If failed to find kind icon, default to tagIcon
          
          typeHandle := IOHandle.fromObj(
            context.rt,
            "fan://notificationExt/resources/alarming/tagIcon.png".toUri
          )
          
          files.add(
            typeHandle.inToBuf.toFile(
              "fan://notificationExt/resources/alarming/${kindName}Icon.png".toUri
            )
          )
        }
      }
    }
    
    return files
  }
  
  protected Void sendEmail(
    HxContext context,
    Obj interest,
    Str[] to)
  {
    // Create email parts
    parts := EmailParts(
      this.generateBody(context, interest),
      this.getAttachments(context, interest)
    )
    
    EmailMod mailLib := context.rt.lib("email")
    FinSmtpMod finSmtp := context.rt.lib("finSmtp")
    
    proposedSubject := (interest is Dict)
      ? Etc.macro(this.subject, interest)
      : ((interest is Grid)
        ? Etc.macro(this.subject, ((Grid) interest).meta)
        : this.subject)
    
    email := email::Email() {
      it.subject = proposedSubject
      it.from = mailLib.settings.from
      it.bcc = to
      it.body = parts.toFinal
    }
    
    token  := finSmtp.getAuthToken()
    client := finSmtp::SmtpClient()

    client.host = mailLib.settings.uri.host
    client.port = mailLib.settings.uri.port ?: 25
    client.username = mailLib.settings.username
    client.password = token ?: mailLib.sys.host.db.passwords.get("email")
    client.oauth = token != null
    client.ssl = mailLib.settings.tls
    client.log = this.log
    client.socketConfig.setTimeouts(SocketConfig.cur.connectTimeout)
        
    client.send(email)
  }
  
  protected Str:TemplateProcessor templateCache()
  {
    return ((Unsafe) this._templateCache.val).val
  }
}

class EmailParts
{
  TextPart emailPart
  FilePart[] files
  
  new make(ContentBody email, File[] attachments)
  {
    this.emailPart = TextPart()
    this.emailPart.headers.set("Content-Type", email.mimeType.toStr)
    this.emailPart.text = email.body
    
    this.files = attachments.map | file | {
      part := FilePart()
      part.file = file
      part.headers.set("Content-Type", "${file.mimeType}; name=${file.basename}")
      part.headers.set("Content-Id", "<${file.basename}>")
      part.headers.set("Content-Disposition", "inline; filename=${file.basename}")
      part.headers.set("Content-Location", file.basename)
      
      return part
    }
  }
  
  public EmailPart toFinal()
  {
    if (this.files.size == 0)
      return this.emailPart
    
    EmailPart[] allParts := [,]
    allParts.add(this.emailPart)
    allParts.addAll(this.files)
    
    multiPart := MultiPart()
    multiPart.parts = allParts
    
    return multiPart
  }
}

const class ContentBody
{
  const Str body
  const MimeType mimeType
  
  new make(Str body, MimeType mimeType)
  {
    this.body = body
    this.mimeType = mimeType
  }
  
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST

using hx
using haystack
using concurrent

const class DirectEmailNotificationTarget : AbstractMailerTarget
{
  public static const Str TYPE := "directEmail"
  
  @TypedTag
  const Str to
      
  new make(Dict rec, |This| f): super(rec, f) {}
      
  public override Future execute(HxContext context, Obj interest)
  {
    this.sendEmail(
      context,
      interest,
      this.to.split(',', true)
    )
    
    return Future.makeCompletable.complete(null)
  }
  
  static new wrap(Dict? dict)
  {
    if (dict == null)
      return null
    
    return create(DirectEmailNotificationTarget#, dict)
  }
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST

using hx
using haystack
using concurrent

const class EmailTopicNotificationTarget : AbstractMailerTarget
{
  public static const Str TYPE := "emailTopic"
  
  @TypedTag
  const Str topic

  const AtomicRef _topicCache := AtomicRef()
  
  new make(Dict rec, |This| f): super(rec, f) {}
  
  public override Future execute(HxContext context, Obj interest)
  {
    if (this.cache == null)
    {
      // Create Cache
      this._topicCache.getAndSet(Unsafe(TopicCache(context, this.topic)))
    }
            
    // Refresh Cache
    try
    {
      cache.refreshIfExpired
      
      users := cache.getImmediateUsersForInterest(context, interest)
      
      if (users.size > 0)
      {
        this.sendEmail(
          context,
          interest,
          users.mapNotNull | user | { user.email }
        )
      }
    }
    catch (Err error)
    {
      this.log.err("Error sending email topic", error)
    }
    
    return Future.makeCompletable.complete(null)
  }
  
  private TopicCache? cache()
  {
    return ((Unsafe?) this._topicCache.val)?.val
  }
  
  static new wrap(Dict? dict)
  {
    if (dict == null)
      return null
    
    return create(EmailTopicNotificationTarget#, dict)
  }
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST

using haystack
using hx
using axon

class NotificationFuncs 
{
  @Axon
  public static Void notifyTarget(Obj target, Obj interest)
  {
    context := HxContext.curHx
    
    NotificationLib lib := context.rt.lib("notification")
    
    if (target is Dict)
      target = ((Dict) target).id
    
    lib.notifyManual(context, target, interest)
  }
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST

using hx
using obs
using concurrent
using haystack

const class NotificationLib : HxLib
{
  public static const Str TARGET_TAG_NAME := "notificationTarget"
  private static const Int RETRY_THRESHOLD := 5 // TODO: Later make configurable as tag on this record
  private static const Duration RETRY_TIME := Duration.fromStr("30sec")
  
  override const NotificationRestAPI web := NotificationRestAPI(this)
  
  private const Unsafe _target := Unsafe([:])
  private const Unsafe _retryQueue := Unsafe([:])
  
  private const ActorPool pool
  private const Actor actor
  
  new make(): super()
  {
    this.pool = ActorPool()
    this.actor = Actor(
      this.pool,
      | NotificationQueueItem item | { this.onHandleNotification(item) }
    )
    
  }
  
  private Ref:NotificationTarget targets()
  {   
    return this._target.val
  }
  
  public override Void onStart()
  {
    this.processTargets
    
    this.observe(
      "obsCommits",
      Etc.makeDict([
        "obsAdds": Marker.val,
        "obsUpdates": Marker.val,
        "obsRemoves": Marker.val,
        "obsFilter": TARGET_TAG_NAME
      ]),
      #noteRecordChangeHandler
    )
    
    super.onStart
  }
  
  public override Void onStop()
  {
    this.pool.stop
    this.pool.join
    super.onStop
  }
  
  protected Obj? onHandleNotification (NotificationQueueItem item)
  {
    if (item.currentTries > RETRY_THRESHOLD)
    {
      this.log.warn(
        "Retrying notifications for $item.interest has exceeded its error threshold."
      )
      
      return null
    }
    
    try
    {
      item.target.execute(
        item.context,
        item.interest
      )
    }
    catch (Err error)
    {
      item.incrementTries
      
      // Add target to retry queue if failed
      this.log.warn("Error exeucting target (attempt $item.currentTries), retrying...", error)
      
      this.actor.sendLater(
        RETRY_TIME,
        item
      )
    }
    
    return null
  }
      
  public Future notify(HxContext context, Str subject, Obj interest)
  {
    // TODO: Make this execute on a separate thread.
    // -- Use coalescing actor to coalesce multiple calls to the same target.
    // -- May need to calculate target
    
    if (interest is Grid && ((Grid) interest).size == 1 && !((Grid) interest).has("digest"))
      interest = ((Grid) interest).get(0)
          
    this.targets.vals.mapNotNull | target | {
      try
      {
        if (target.include(context, interest))
        {
          if (interest is Dict)
            interest = [interest]
          
          return NotificationQueueItem(context, interest, target)
        }
      }
      catch (Err error)
      {
        // Error executing target include
        this.log.warn("Error testing include for notification target", error)
      }     
      
      return null
    }.each | item | {
      actor.send(item)
    }
    
    // Send notification to any connected web sockets.
    this.web.sendNotification(subject, interest)
    
    return Future.makeCompletable.complete(null)
  }
  
  public Future notifyManual(HxContext context, Ref targetRef, Obj interest)
  {
    try
    {
      target := this.targets.get(targetRef)
      return target.execute(context, interest)
    }
    catch (Err err)
    {
      return Future.makeCompletable.completeErr(err)
    }
  }
  
  private Ref:NotificationQueueItem retryQueue()
  {
    return this._retryQueue.val
  } 
    
  private Void noteRecordChangeHandler(CommitObservation observation)
  {
    switch (observation.subType)
    {
      //case CommitObservationAction.added.name:
      //  this.processTarget(observation.newRec)
      
      case CommitObservationAction.updated.name:
        this.targets.remove(observation.id)
      
      case CommitObservationAction.removed.name:
        this.targets.remove(observation.id)
    }
    
    this.processTarget(observation.newRec)
  }
  
  private Void processTargets()
  {
    targets := this.rt.db.readAllList(
      Filter.fromStr(TARGET_TAG_NAME)
    )
    
    targets.each | target | {
      this.processTarget(target)
    }
  }
  
  private Void processTarget(Dict dict)
  {
    type := TARGET_REGISTRY.get(dict.get(TARGET_TAG_NAME))
    
    constructorMethod := type.method("wrap", false)
    
    if (constructorMethod != null)
    {
      targetInst := constructorMethod.call(dict) as NotificationTarget
      this.targets.set(dict.id, targetInst)   
    }
  }
  
  
  private static const Str:Type TARGET_REGISTRY := [
    DirectEmailNotificationTarget.TYPE: DirectEmailNotificationTarget#,
    EmailTopicNotificationTarget.TYPE: EmailTopicNotificationTarget#,
    PushBulletNotificationTarget.TYPE: PushBulletNotificationTarget#
  ]
}

const class NotificationQueueItem
{
  private const AtomicRef _interests := AtomicRef(Unsafe([,]))
  public const NotificationTarget target
  private const AtomicInt tryCount := AtomicInt(0)
  private const Unsafe _context
  
  
  new make (HxContext context, Obj[] interest, NotificationTarget target)
  {
    this.interestInternal.addAll(interest)
    this.target = target
    this._context = Unsafe(context)
  }
  
  public Void incrementTries()
  {
    this.tryCount.add(1)
  }
  
  public Int currentTries()
  {
    return this.tryCount.val
  }
  
  public HxContext context()
  {
    return this._context.val
  }
  
  private Obj[] interestInternal()
  {
    return ((Unsafe)this._interests.val).val
  }
  
  public Obj interest()
  {
    if (this.interestInternal.size == 1)
      return this.interestInternal.get(0)
    
    return this.interestInternal
  }
  
  public Void addInterests(Obj[] interests)
  {
    this.interestInternal.addAll(interests)
  }
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST

using hx
using web
using util
using haystack

const class NotificationRestAPI : HxLibWeb 
{
  const Unsafe _sockets := Unsafe([,])
  
  override NotificationLib lib()
  {
    return super.lib
  }
  
  new make(NotificationLib lib): super(lib)
  {
  }
  
  private WebSocket[] sockets()
  {
    return this._sockets.val
  }
  
  override Void onGet()
  {
    operation := this.req.modRel.path.getSafe(0, "default")
    context := this.rt.user.authenticate(req, res)
    
    if (operation == "ws")
    {
      server := WebSocket.openServer(req, res)
      this.sockets.add(server)
    }
  }
  
  public Void sendNotification(Str subject, Obj notification)
  {
    if (this.sockets.size < 1)
      return
    
    closedSockets := [,]
    
    // Encode notification to hayson
    dataStr := StrBuf()
    writer := JsonWriter(dataStr.out)
    writer.writeVal(notification)
    writer.close
    
    output := dataStr.toStr
    
    // Send data to each connected socket
    this.sockets.each | socket | 
    {
      if (socket.isClosed)
      {
        closedSockets.add(socket)
        return
      }
      
      try
      {
        socket.send(
          output
        )
      }
      catch (Err error)
      {
        socket.close()
        closedSockets.add(socket)
      }
    }
    
    // Remove sockets that have closed
    closedSockets.each | socket | {  
      this.sockets.remove(socket)
    }
  }
  
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST

using haystack
using hx
using concurrent

abstract const class NotificationTarget : TypedDict
{
  protected const Log log := Log.get("NotificationTarget-${typeof.name}")
  // Notification Type
  @TypedTag
  const Str notificationTarget
  
  // Notification filter. This filter should select the alarm conditions and sourceRef->tags
  @TypedTag
  const Str? notificationOn
  
  @TypedTag
  const Bool disabled
    
  const Filter? onFilter
  
  new make(Dict rec, |This| f): super(rec)
  {
    f(this)
    
    if (this.notificationOn != null)
      this.onFilter = Filter.fromStr(this.notificationOn)
  }
  
  public abstract Future execute(HxContext context, Obj interest)
  
  public Bool include(HxContext context, Obj interest)
  {
    if (this.disabled)
      return false
    
    if (this.onFilter != null)
    {
      if (interest is List)
      {
        return ((List) interest).all | item | {  
          this.onFilter.matches(item, context)
        }
      }
      
      if (interest is Grid)
      {
        return ((Grid) interest).all | row | {
          this.onFilter.matches(row, context)
        }
      }

      if (interest is Dict)
        return this.onFilter.matches(interest, context)
      
      return false
    }
    
    return true
  }
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST

using concurrent
using haystack
using hx

const class PushBulletNotificationTarget : NotificationTarget
{
  public static const Str TYPE := "pushBullet"
  
  new make(Dict rec, |This| f): super(rec, f) {}
  
  public override Future execute(HxContext context, Obj interest)
  {
    return Future.makeCompletable.completeErr(Err("Not implemented"))
  }
  
  static new wrap(Dict? dict)
  {
    if (dict == null)
      return null
    
    return create(PushBulletNotificationTarget#, dict)
  }
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST

using haystack
using hx
using skyarcd

class TopicCache 
{
  static const Duration CACHE_PERIOD := 27sec
  
  const Str topicName
  
  private const Log log
  private HxContext context
  private Ref:Topic topics
  private Int lastRefreshed // Last Refreshed Ticks
  
  new make(HxContext context, Str topicName)
  {
    this.context = context
    this.topicName = topicName
    this.log = Log.get("TopicCache")
    this.topics = [:]
  }
  
  public Bool isExpired()
  {
    return (Duration.nowTicks - this.lastRefreshed) > CACHE_PERIOD.ticks
  }
  
  public Void refreshIfExpired()
  {
    if (this.isExpired)
      this.refresh()
  }
  
  private Void refresh()
  {
    this.lastRefreshed = Duration.nowTicks
    
    // Get Topics
    this.topics.clear
    
    topics := context.db.readAllList(
      Filter.fromStr("${this.topicName}Topic and topicOn")
    ).map | rec | { 
      Topic.wrap(rec) 
    }
    
    topics.each | Topic topic | { 
      this.topics.set(topic.id, topic) 
    } 
  
    // Populate users in topic
    UserMod userMod := context.rt.lib("user")
    
    userMod.users.each | user | 
    {
      if (user.isDisabled || user.email == null)
        return
      
      // Populate Topics to send immediately
      if (user.has("topicsEmailImmediate"))
      {
        try
        {
          topicIds := Ref.listFromStr(
            user.meta.get("topicsEmailImmediate")
          )
          
          topicIds.each | topicId | {  
            this.topics.getChecked(topicId, false)?.addImmediate(EmailUser(user))
          }
        }
        catch (Err err)
        {
          // Invalid Immediate Ids
          log.err("Invalid immediate ids", err)
        }
      }
      
      // Populate topics to send as a digest
      if (user.has("topicsEmailDigest"))
      {
        try
        {
          topicIds := Ref.listFromStr(
            user.meta.get("topicsEmailDigest")
          )
          
          topicIds.each | topicId | {  
            this.topics.getChecked(topicId, false)?.addDigest(EmailUser(user))
          }
        }
        catch (Err err)
        {
          // Invalid Digest Ids
          log.err("Invalid digest ids", err)
        }
      }
    }
  }
  
  public HxUser[] getImmediateUsersForInterest(HxContext context, Obj interest)
  {
    // TODO: Ensure the email isn't in the list twice.
    Dict[]? interests
    
    if (interest is Grid)
      interests = ((Grid) interest).toRows
    else
    if (interest is List)
      interests = interest
    if (interest is Dict)
      interests = [interest]
    
    
    Topic[] matchedTopics := interests.flatMap | item | {
      return this.topics.vals.findAll | topic | {
        topic.on.matches(item, context)
      }
    }   
    
    currentTime := DateTime.now.time
    
    return matchedTopics.flatMap | topic | {  
      topic.immediateUsers
        .mapNotNull | user| {  
          if (user.shouldSilenceEmail(currentTime))
            return null
          
          return user.user
        }
    }
  }
  
  public HxUser[] getDigestUsersForInterest(HxContext context, Obj interest)
  {
    Dict[]? interests
    
    if (interest is Grid)
      interests = ((Grid) interest).toRows
    else
    if (interest is List)
      interests = interest
    if (interest is Dict)
      interests = [interest]
    
    
    Topic[] matchedTopics := interests.flatMap | item | {
      return this.topics.vals.findAll | topic | {
        topic.on.matches(item, context)
      }
    }
    
    currentTime := DateTime.now.time
    
    return matchedTopics.flatMap | topic | {  
      topic.digestUsers
        .mapNotNull | user| {  
          if (user.shouldSilenceEmail(currentTime))
            return null
          
          return user.user
        }
    }
  }
}

const class Topic : TypedDict
{
  @TypedTag
  const Str topicOn
  
  const Filter on
  
  private const Unsafe _immediateUsers
  private const Unsafe _digestUsers
  
  new make(Dict rec, | This | f): super(rec)
  {
    f(this)
    
    this.on = Filter.fromStr(this.topicOn)
    this._immediateUsers = Unsafe([,])
    this._digestUsers = Unsafe([,])
  }
  
  public EmailUser[]? immediateUsers()
  {
    return this._immediateUsers.val
  }
  
  public EmailUser[]? digestUsers()
  {
    return this._immediateUsers.val
  }
  
  public Void addImmediate(EmailUser user)
  {
    this.immediateUsers.add(user)
  }
  
  public Void addDigest(EmailUser user)
  {
    this.digestUsers.add(user)
  }
  
  static new wrap(Dict dict)
  {
    return create(Topic#, dict, | err | { echo("Error creating topic: $err") })
  }
}

const class EmailUser
{
  const HxUser user
  const Time? silenceStartTime
  const Time? silenceEndTime
  
  new make(HxUser user)
  {
    this.user = user
    
    if (user.meta.has("emailSilenceStartTime"))
      this.silenceStartTime = user.meta.get("emailSilenceStartTime")
    
    if (user.meta.has("emailSilenceEndTime"))
      this.silenceEndTime = user.meta.get("emailSilenceEndTime")
  }
  
  public Bool shouldSilenceEmail(Time when)
  {
    if (this.silenceStartTime == null)
      return false
    
    isSilenced := false
    
    if (when.compare(this.silenceStartTime) == 1)
    {
      isSilenced = true
      
      if (this.silenceEndTime != null)
      {
        if (when.compare(this.silenceEndTime) == 1)
          isSilenced = false
      }
    }
    
    
    return isSilenced
  }
}

Skyspark 3.1.12 ∙ 20-Oct-2026 14:19 CEST