engine

type BrokerPort interface {
	// Publish sends a server-side message (docs/ROADMAP.md §6 file 5.7):
	// retain for properties, per-message expiry for datastreams.
	Publish(topic string, payload []byte, qos byte, retain bool, expiry time.Duration) error
	// RefreshIntrospection reloads a connected device's introspection-derived
	// ACL state after the engine persists a new introspection
	// (docs/ROADMAP.md §7.2 file 6.7).
	RefreshIntrospection(ctx context.Context, realm string, id deviceid.ID) error
}

type Config struct {
	// Shards is the number of ordered pipeline shards (default
	// DefaultShards). Messages of one device always land on the same shard.
	Shards int
	// ShardQueue is the per-shard channel capacity (default
	// DefaultShardQueue). A full shard blocks QoS >= 1 submits (deferred-ack
	// backpressure) and drops QoS 0 messages with a metric (§1.4).
	ShardQueue int
	// BatchMaxRows is the micro-batch row cap (default DefaultBatchMaxRows).
	BatchMaxRows int
	// BatchMaxWait is the micro-batch time cap (default DefaultBatchMaxWait).
	BatchMaxWait time.Duration
	// MaxPayloadBytes caps accepted data payload size for both formats
	// (default payload.DefaultMaxSize).
	MaxPayloadBytes int
	// Forwarder receives trigger actions that are not HTTP webhooks
	// (upstream's AMQP actions, docs/DESIGN.md §1.1). nil is the default and
	// means those actions are logged and counted "skipped", which is the
	// designed behaviour rather than a gap.
	Forwarder triggers.Forwarder
	// Registerer receives the engine's Prometheus collectors; nil leaves
	// them unregistered (the collectors still work, which tests rely on).
	Registerer prometheus.Registerer
	// Logger receives engine logs (default slog.Default()).
	Logger *slog.Logger
}

type Engine struct {
	// contains filtered or unexported fields
}

type OpKind uint8

type PersistOp struct {
	// Kind discriminates the persistence action.
	Kind OpKind
	// Realm and RealmID identify the tenant.
	Realm   string
	RealmID int16
	// DeviceID is the publishing device.
	DeviceID deviceid.ID
	// Interface is the compiled interface the message validated against.
	Interface *interfaceschema.CompiledInterface
	// Mapping is the matched endpoint mapping; nil for object aggregation.
	Mapping *interfaceschema.CompiledMapping
	// Path is the concrete data path ("" only for flat object aggregation).
	Path string
	// Value is the decoded payload value (payload.Value closed set;
	// map[string]payload.Value for objects; nil for property unset).
	Value payload.Value
	// Format is the wire format the payload arrived in.
	Format payload.Format
	// TS is the effective sample timestamp: the explicit `t` when the
	// mapping declares explicit_timestamp, broker reception time otherwise.
	TS time.Time
	// ReceptionTS is the broker reception timestamp.
	ReceptionTS time.Time

	// Previous-value snapshot for the change-derived trigger conditions,
	// captured at accept time by trackPrevious when a trigger is actually
	// watching this endpoint: Prev holds the canonical §2.3 JSON of the
	// previous value, PrevExists reports whether there was one, Tracked
	// marks that a snapshot was taken at all (untracked ops skip change
	// events but still fire incoming_data).
	Tracked    bool
	PrevExists bool
	Prev       []byte
	// contains filtered or unexported fields
}

type Store interface {
	// ListRealms feeds the realm name/ID resolution in the schema cache.
	ListRealms(ctx context.Context) ([]store.Realm, error)
	// LoadRealmInterfaces returns every installed interface of a realm for
	// schema-snapshot rebuilds.
	LoadRealmInterfaces(ctx context.Context, realmID int16) ([]*store.StoredInterface, error)
	// Listen subscribes to a NOTIFY channel (cache invalidation,
	// docs/DESIGN.md §2.6).
	Listen(ctx context.Context, channel string) (<-chan store.Notification, error)
	// GetDevice loads a device's introspection and payload-format hint.
	GetDevice(ctx context.Context, realmID int16, id deviceid.ID) (*store.Device, error)
	// SetPayloadFormatHint persists the sticky payload-format flip
	// (docs/DESIGN.md §3.5.4).
	SetPayloadFormatHint(ctx context.Context, realmID int16, id deviceid.ID, hint string) error
	// AppendDatastreams commits one micro-batch in a single transaction.
	AppendDatastreams(ctx context.Context, batch store.DatastreamBatch) error
	// UpsertProperty applies a property set (last-value-wins).
	UpsertProperty(ctx context.Context, p store.Property) error
	// UnsetProperty applies a property unset (row delete).
	UnsetProperty(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64, path string) (bool, error)
	// GetProperty reads one stored property — the previous-value lookup for
	// property change triggers.
	GetProperty(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64, path string) (*store.Property, error)
	// LatestIndividual reads the newest sample of one datastream series —
	// the previous-value lookup for datastream change triggers.
	LatestIndividual(ctx context.Context, realmID int16, deviceID deviceid.ID, interfaceID int64, path string) (*store.IndividualRow, error)
	// UpdateIntrospection replaces a device's introspection, returning the
	// (name, major) pairs that were dropped (docs/ROADMAP.md §7.2 file 6.7).
	UpdateIntrospection(ctx context.Context, realmID int16, id deviceid.ID, intro map[string]store.InterfaceVersion) (map[string]store.InterfaceVersion, error)
	// PurgeDeviceOwnedExcept implements the `producer/properties` resync
	// (docs/DESIGN.md §3.3): device-owned properties not in keep are deleted.
	PurgeDeviceOwnedExcept(ctx context.Context, realmID int16, deviceID deviceid.ID, keep []store.PropertyRef) (int64, error)
	// ListServerOwnedProperties returns the device's server-owned properties
	// (emptyCache resend + `consumer/properties` payload, docs/DESIGN.md §3.4).
	ListServerOwnedProperties(ctx context.Context, realmID int16, deviceID deviceid.ID) ([]store.Property, error)
	// ListTriggers returns a realm's installed triggers for the compiled
	// trigger cache (docs/ROADMAP.md §7.2 file 6.10).
	ListTriggers(ctx context.Context, realmID int16) ([]store.Trigger, error)
	// ListTriggerPolicies returns every delivery policy of a realm so the
	// snapshot can attach compiled policies to triggers at load time.
	ListTriggerPolicies(ctx context.Context, realmID int16) ([]store.TriggerPolicy, error)
}