Stream Reporting Dashboard

/**
 * Demonstrates persistence and stream queries.
 * One workflow logs events, another queries and reports on them.
 */

webhook log_event {
  label: "Log Event"
  schema: @json {
    {
      "type": "object",
      "required": ["event_type", "value"],
      "properties": {
        "event_type": { "type": "string" },
        "value": { "type": "number" },
        "source": { "type": "string" }
      }
    }
  }
}

schedule daily_report {
  label: "Daily Report"
  cron: "0 17 * * *"
  timezone: "UTC"
}

workflow record_event {
  label: "Record Event"

  root {
    type: code
    label: "Store event"
    code: @ts { return context.nodes.root.input }
    outputSchema: @json {
      {
        "type": "object",
        "properties": {
          "event_type": { "type": "string" },
          "value": { "type": "number" },
          "source": { "type": "string" }
        }
      }
    }
  }
}

workflow generate_report {
  label: "Generate Daily Report"

  root {
    type: code
    label: "Start"
    code: @ts { return { date: new Date().toISOString().split("T")[0] } }
    outputSchema: @json {
      {
        "type": "object",
        "properties": {
          "date": { "type": "string" }
        }
      }
    }
  }

  node recent_events {
    type: stream
    label: "Query recent events"
    stream: events
    version: v1
    filter: @ts {
      const since = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString()
      return { created_at: { gte: since } }
    }
  }

  node summarize {
    type: code
    label: "Summarize events"
    code: @ts {
      const events = context.nodes.recent_events.output || []
      const byType = {}
      for (const event of events) {
        const t = event.event_type || "unknown"
        if (!byType[t]) byType[t] = { count: 0, total_value: 0 }
        byType[t].count += 1
        byType[t].total_value += event.value || 0
      }
      return {
        date: context.nodes.root.output.date,
        total_events: events.length,
        by_type: byType
      }
    }
  }

  node send_report {
    type: email
    label: "Send report"
    from: @ts { return "[email protected]" }
    to: @ts { return "[email protected]" }
    subject: @ts { return "Daily Event Report - " + context.nodes.root.output.date }
    text: @ts {
      const summary = context.nodes.summarize.output
      return "Daily Event Report\n\nTotal events: " + summary.total_events + "\n\nBreakdown:\n" + JSON.stringify(summary.by_type, null, 2)
    }
  }

  flow {
    root -> recent_events
    recent_events -> summarize
    summarize -> send_report
  }
}

stream events {
  label: "Events"
  workflow: record_event
  version: v1

  versions: {
    v1 {
      schema: @json {
        {
          "type": "object",
          "required": ["event_type", "value"],
          "properties": {
            "event_type": { "type": "string" },
            "value": { "type": "number" },
            "source": { "type": "string" }
          }
        }
      }

      condition: @ts { return context.output.root != null }

      prepare: @ts { return context.output.root }
    }
  }
}

trigger on_event {
  webhook:log_event -> record_event
  enabled: true
}

trigger on_daily_report {
  schedule:daily_report -> generate_report
  enabled: true
}