Documentation
¶
Overview ¶
Package event implements the DingTalk Stream event subscription pipeline for dws. The architecture is a single-cloud-connection bus daemon (one per ClientID) plus N local consumer processes communicating over Unix socket / Windows Named Pipe. The bus keeps one cloud connection per identity while exposing observable connection state, per-event-type metrics, and Hello-time filter pushdown to local consumers.
Package layout:
event/ // top-level types (RawEvent, EmitFn, hash helpers) event/dedup/ // event_id LRU dedup event/registry/ // catch-all event types + compact processor registry event/source/ // wrap dingtalk-stream-sdk-go + connection state machine event/bus/ // daemon loop, hub, metrics, lockfile, meta event/transport/ // UDS/Pipe abstraction, frame protocol event/busctl/ // discover, spawn, stop helpers event/consume/ // consumer-side pipeline, formatter, router, sink event/lock/ // cross-platform flock primitive (Unix flock / Windows LockFileEx) event/process/ // cross-platform process-alive check (Unix signal 0 / Windows OpenProcess)
See plans/2026-05-28_event_capability_v1.plan.md for the full design, invariants, and protocol.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ClientIDHash ¶
ClientIDHash returns the path-safe identifier derived from the (possibly unsafe) ClientID string. We use sha256(clientID)[:8] hex-encoded (16 chars); this is the value placed in filesystem paths and Named Pipe names. 8 bytes (64 bits) is sufficient to distinguish a handful of ClientIDs on one machine while keeping the Unix socket path under macOS's 104-byte sun_path limit. The original ClientID is stored in bus.meta and shown in status output.
func IPCEndpoint ¶
func IPCEndpoint(workDir, editionName string, sourceKind SourceKind, identityHash string) string
IPCEndpoint returns the bus IPC endpoint for one identity: a Named Pipe name on Windows, otherwise bus.sock inside workDir.
The canonical Unix location is <workDir>/bus.sock, but workDir derives from the config dir, which can be arbitrarily deep (e.g. dwssb sandboxes use ~/.dwssb/sandboxes/<name>/config/...). When the canonical path would exceed the OS sun_path limit, the socket falls back to a short deterministic path under os.TempDir keyed by a hash of workDir, so every process (consume parent, forked _bus child, status/stop tooling) that derives the endpoint from the same workDir agrees on the location. bus.lock / bus.meta / bus.log always stay in workDir — only the socket moves.
This is the single source of truth for endpoint derivation; the cobra layer and busctl must not re-implement the shape.
func IdentityHash ¶
IdentityHash returns the path-safe identifier derived from any event identity string. It intentionally uses the same algorithm as ClientIDHash so existing app-stream paths remain stable when the identity is client_id.
func MaxUnixSocketPath ¶
func MaxUnixSocketPath() int
MaxUnixSocketPath returns the longest Unix socket path accepted by bind/connect on this OS (Go rejects longer names with EINVAL before the syscall). sockaddr_un.sun_path is 104 bytes on darwin and the BSDs and 108 on Linux; the usable budget is one less.
func RedactSecret ¶
RedactSecret returns a redacted form of a secret string for logging and error messages: first 3 + "***" + last 3 characters. Short secrets (<= 6 chars) are fully masked. Empty input yields empty output.
Types ¶
type EmitFn ¶
type EmitFn func(*RawEvent)
EmitFn is the non-blocking handoff from Source to Hub. Implementations MUST return immediately (drop-oldest on the consumer's sendCh) so the SDK's callback can ACK without delay — see plan invariant #1.
type RawEvent ¶
type RawEvent struct {
EventID string `json:"event_id"` // DataFrameHeader "eventId"
EventBornTime int64 `json:"event_born_time"` // milliseconds; verified P0
EventCorpID string `json:"event_corp_id"` // tenant corp id
EventType string `json:"event_type"` // catch-all/filter routing key
EventUnifiedAppID string `json:"event_unified_app_id"` // app id
EventScope string `json:"event_scope,omitempty"`
SubscribeID string `json:"subscribe_id,omitempty"`
SourceID string `json:"source_id,omitempty"`
RuleType string `json:"rule_type,omitempty"`
Data string `json:"data"` // raw JSON payload from SDK
Headers map[string]string `json:"headers,omitempty"` // full DataFrame.Headers, passthrough
ReceivedAt time.Time `json:"received_at"` // bus receive time (UTC)
}
RawEvent is one event delivered by the Source layer to the Hub. It mirrors the DingTalk Stream SDK's EventHeader (5 fields) plus the raw JSON payload from DataFrame.Data, plus the local receive timestamp used for ordering and dedup fallback keys.
Field naming follows Go conventions (camelCase capitalised); JSON tags follow the SDK's wire format so that NDJSON output is consistent with what DingTalk documents on its open platform.
type SourceKind ¶
type SourceKind string
SourceKind identifies the upstream event connection family.
const ( SourceKindAppStream SourceKind = "app_stream" SourceKindPersonalStream SourceKind = "personal_stream" )
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package bus implements the daemon side of the dws event subsystem: one long-lived process per ClientID that holds the single cloud connection and fans out events to N local consumers over IPC.
|
Package bus implements the daemon side of the dws event subsystem: one long-lived process per ClientID that holds the single cloud connection and fans out events to N local consumers over IPC. |
|
Package busctl glues the consume client to the bus daemon.
|
Package busctl glues the consume client to the bus daemon. |
|
Package consume implements the consumer-side process of `dws event consume`: dial the bus, send Hello, read Event frames, format them, and write them out (stdout / file / dir).
|
Package consume implements the consumer-side process of `dws event consume`: dial the bus, send Hello, read Event frames, format them, and write them out (stdout / file / dir). |
|
Package dedup implements a fixed-capacity LRU set used by the bus to suppress duplicate events that the DingTalk Stream SDK redelivers on reconnect (see plan invariant #2).
|
Package dedup implements a fixed-capacity LRU set used by the bus to suppress duplicate events that the DingTalk Stream SDK redelivers on reconnect (see plan invariant #2). |
|
Package lock implements a cross-platform exclusive file lock primitive used by the bus daemon to enforce the "single bus per ClientID" invariant (plan invariant #3).
|
Package lock implements a cross-platform exclusive file lock primitive used by the bus daemon to enforce the "single bus per ClientID" invariant (plan invariant #3). |
|
Package process provides cross-platform process-existence checks.
|
Package process provides cross-platform process-existence checks. |
|
Package registry holds two cross-cutting reference data sets used by daemon, consume, and the cobra command layer:
|
Package registry holds two cross-cutting reference data sets used by daemon, consume, and the cobra command layer: |
|
Package source wraps the open-dingtalk Stream SDK and exposes a small blocking Start interface plus a connection state machine.
|
Package source wraps the open-dingtalk Stream SDK and exposes a small blocking Start interface plus a connection state machine. |
|
Package transport implements the cross-platform local IPC between the bus daemon and consume clients.
|
Package transport implements the cross-platform local IPC between the bus daemon and consume clients. |