// Package mqtt is the Control Service's link to the broker (AGENTS.md §10 // rules 5–6, §11 Phase 3; TRANSPORT_DESIGN.md). // // The broker is transport only. This package publishes each device's desired // state, retained, on restrict//config, and consumes the reports that // devices write back. SQLite stays the source of truth (SYSTEM_MAP.md): nothing // here is authoritative, and a report is handed straight to the store. // // Topics (TRANSPORT_DESIGN.md "Topic design"): // // restrict//config service → device, retained // restrict//status device → service, retained (presence) // restrict//telemetry device → service, not retained // // "config" keeps the exact topic the device table already stores, so a device's // identity never changes when it is renamed. package mqtt import ( "encoding/json" "fmt" "strings" "time" "gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store" ) // TopicPrefix is the first topic level for every message this service touches. const TopicPrefix = "restrict" // Topic suffixes, appended to "restrict/". const ( SuffixConfig = "/config" SuffixStatus = "/status" SuffixTelemetry = "/telemetry" ) // StatusSubscription and TelemetrySubscription are the wildcard topics the // Control Service reads. A device may write only its own status and telemetry // (AGENTS.md §10 rule 6). const ( StatusSubscription = TopicPrefix + "/+/status" TelemetrySubscription = TopicPrefix + "/+/telemetry" ) // StatusTopic returns the status topic for a device's config topic. func StatusTopic(configTopic string) string { return withSuffix(configTopic, SuffixStatus) } // TelemetryTopic returns the telemetry topic for a device's config topic. func TelemetryTopic(configTopic string) string { return withSuffix(configTopic, SuffixTelemetry) } func withSuffix(configTopic, suffix string) string { return strings.TrimSuffix(configTopic, SuffixConfig) + suffix } // deviceFromTopic turns "restrict/harry_phone_2/status" back into the config // topic "restrict/harry_phone_2/config" so the store can find the device. func deviceFromTopic(topic, suffix string) (string, bool) { if !strings.HasPrefix(topic, TopicPrefix+"/") || !strings.HasSuffix(topic, suffix) { return "", false } base := strings.TrimSuffix(topic, suffix) if strings.Count(base, "/") != 1 || len(base) <= len(TopicPrefix)+1 { return "", false } return base + SuffixConfig, true } // Config is the desired state for one device. It is published retained on // restrict//config: a device that reconnects is configured at once, // with no poll and no waiting (TRANSPORT_DESIGN.md). // // The shape is versioned with "v" and is consumed by the kids' clients in // Phases 4 and 5. Add fields; do not repurpose them. type Config struct { V int `json:"v"` DeviceID int64 `json:"device_id"` Name string `json:"name"` Platform string `json:"platform"` Locked bool `json:"locked"` Until *time.Time `json:"until"` Allow []string `json:"allow"` Block []string `json:"block"` // Reason is why the device is locked: "schedule" for an active weekly // window, "manual" for a lock from the dashboard or the admin API. It is // omitted when the device is not locked. It is additive (Phase 6), so v // stays 1. Reason string `json:"reason,omitempty"` // GrantUntil is set when a time-limited exemption is active (Phase 8): the // device is unlocked and must not re-lock before this instant. It is omitted // when there is no grant. Additive, so v stays 1. GrantUntil *time.Time `json:"grant_until,omitempty"` GeneratedAt time.Time `json:"generated_at"` } // ConfigVersion is the current schema version of Config. const ConfigVersion = 1 // Marshal renders the payload. Lists are always present, never null, so a // client can parse without null checks. func (c Config) Marshal() ([]byte, error) { if c.Allow == nil { c.Allow = []string{} } if c.Block == nil { c.Block = []string{} } b, err := json.Marshal(c) if err != nil { return nil, fmt.Errorf("marshal config: %w", err) } return b, nil } // Status is a presence report from a device. // // The device publishes a retained {"online": true} when it comes up and // registers the will {"online": false} on the same topic, so the broker // publishes "offline" the moment the connection drops without a clean // goodbye. That is the whole of presence detection (FR-6). type Status struct { Online *bool `json:"online"` App string `json:"app"` Locked *bool `json:"locked"` At string `json:"at"` } // ParseStatus reads a status payload. A report with no "online" field means the // device is talking, so it counts as online. func ParseStatus(payload []byte) (Status, error) { var st Status if err := json.Unmarshal(payload, &st); err != nil { return Status{}, fmt.Errorf("parse status: %w", err) } if len(payload) == 0 { return Status{}, fmt.Errorf("parse status: empty payload") } return st, nil } // IsOnline reports the presence the payload asserts. func (s Status) IsOnline() bool { return s.Online == nil || *s.Online } // At parses the report timestamp, falling back to the current time. func (s Status) Timestamp(now time.Time) time.Time { if s.At == "" { return now } for _, layout := range []string{time.RFC3339, time.RFC3339Nano} { if t, err := time.Parse(layout, s.At); err == nil { return t } } return now } // UsageEntry is one app or domain total. type UsageEntry struct { Name string `json:"name"` Seconds int `json:"seconds"` } // Telemetry is a rolling daily usage report from a device. Devices report // totals for the day, not deltas, so a repeated report replaces the row // (migration 002). // // Two shapes are accepted: the list form // // {"day":"2026-10-13","apps":[{"name":"com.spotify.music","seconds":600}], // "domains":[{"name":"youtube.com","seconds":120}]} // // and the flat single-row form // // {"app":"com.spotify.music","seconds":600} type Telemetry struct { Day string `json:"day"` Apps []UsageEntry `json:"apps"` Domains []UsageEntry `json:"domains"` App string `json:"app"` Domain string `json:"domain"` Seconds int `json:"seconds"` } // ParseTelemetry reads a telemetry payload. func ParseTelemetry(payload []byte) (Telemetry, error) { var t Telemetry if len(payload) == 0 { return Telemetry{}, fmt.Errorf("parse telemetry: empty payload") } if err := json.Unmarshal(payload, &t); err != nil { return Telemetry{}, fmt.Errorf("parse telemetry: %w", err) } return t, nil } // DayOrDefault returns the day the report belongs to, in the given location. A // device may send its own local day; anything that is not YYYY-MM-DD is // ignored and today's date is used instead. func (t Telemetry) DayOrDefault(now time.Time) string { if t.Day != "" { if d, err := time.Parse("2006-01-02", t.Day); err == nil { return d.Format("2006-01-02") } } return now.Format("2006-01-02") } // Rows flattens the payload into telemetry rows. The app and the domain are // kept in separate columns; a row with neither is dropped. func (t Telemetry) Rows() []store.TelemetryEntry { out := make([]store.TelemetryEntry, 0, len(t.Apps)+len(t.Domains)+1) for _, e := range t.Apps { if e.Name == "" || e.Seconds < 0 { continue } out = append(out, store.TelemetryEntry{App: e.Name, Seconds: e.Seconds}) } for _, e := range t.Domains { if e.Name == "" || e.Seconds < 0 { continue } out = append(out, store.TelemetryEntry{Domain: e.Name, Seconds: e.Seconds}) } if (t.App != "" || t.Domain != "") && t.Seconds >= 0 { out = append(out, store.TelemetryEntry{App: t.App, Domain: t.Domain, Seconds: t.Seconds}) } return out }