// Package notifier handles alerts and approval requests via Matrix. // Uses the DB as the rendezvous — no service-to-service calls (SA7/A7). // Pending approvals survive restarts of either side. package notifier import ( "bytes" "context" "crypto/hmac" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "log/slog" "net/http" "strings" "time" "github.com/dtoro/oikos/internal/config" "github.com/dtoro/oikos/internal/db" "github.com/dtoro/oikos/internal/health" "github.com/google/uuid" ) // Run starts the notifier loop. Blocks until ctx is cancelled. func Run(ctx context.Context, pool *db.Pool, cfg config.Config) { slog.Info("notifier: starting") // Liveness probe (plan D5): the notifier ticks every 15s (approvals) and // 30s (reactions). 2 min staleness covers a slow Matrix round-trip plus a // missed tick without false-failing. probe := health.New(2 * time.Minute) probe.Serve(ctx, cfg.HealthListen) processPendingApprovals(ctx, pool, cfg) probe.Bump() ticker := time.NewTicker(15 * time.Second) defer ticker.Stop() reactionTimer := time.NewTicker(30 * time.Second) defer reactionTimer.Stop() for { select { case <-ctx.Done(): slog.Info("notifier: shutting down") return case <-ticker.C: processPendingApprovals(ctx, pool, cfg) probe.Bump() case <-reactionTimer.C: pollReactions(ctx, pool, cfg) probe.Bump() } } } // RunnerForMain provides the run function for registration in main. func RunnerForMain() func(context.Context, *db.Pool, config.Config) { return Run } type pendingApproval struct { ID uuid.UUID Action string RiskClass string TokenHash *string AlertSentAt *time.Time MatrixEventID *string ExpiresAt time.Time } // processPendingApprovals finds pending approvals, generates tokens, and sends Matrix alerts. func processPendingApprovals(ctx context.Context, pool *db.Pool, cfg config.Config) { rows, err := pool.Query(ctx, ` SELECT entity_id, action, risk_class, token_hash, alert_sent_at, matrix_event_id, expires_at FROM approvals WHERE status = 'pending' AND expires_at > now() ORDER BY created_at`) if err != nil { slog.Error("notifier: list approvals", "error", err) return } defer rows.Close() var approvals []pendingApproval for rows.Next() { var a pendingApproval if err := rows.Scan(&a.ID, &a.Action, &a.RiskClass, &a.TokenHash, &a.AlertSentAt, &a.MatrixEventID, &a.ExpiresAt); err != nil { slog.Error("notifier: scan approval", "error", err) continue } approvals = append(approvals, a) } for _, a := range approvals { if a.ExpiresAt.Before(time.Now()) { pool.Exec(ctx, "UPDATE approvals SET status = 'expired' WHERE entity_id = $1", a.ID) continue } if a.TokenHash == nil || *a.TokenHash == "" { token := generateApprovalToken(a.ID, cfg.ApprovalHMACSecret) tokenHash := hashToken(token) pool.Exec(ctx, "UPDATE approvals SET token_hash = $2 WHERE entity_id = $1", a.ID, tokenHash) a.TokenHash = &tokenHash } if a.AlertSentAt != nil { continue } eventID, err := sendMatrixAlert(ctx, cfg, a.ID, a.Action, a.RiskClass) if err != nil { slog.Error("notifier: send Matrix alert", "error", err, "approval_id", a.ID) continue } pool.Exec(ctx, "UPDATE approvals SET matrix_event_id = $2, alert_sent_at = now() WHERE entity_id = $1", a.ID, eventID) } } // pollReactions checks Matrix for ✅/❌ reactions on sent approval messages. func pollReactions(ctx context.Context, pool *db.Pool, cfg config.Config) { if cfg.MatrixHomeserver == "" || cfg.MatrixToken == "" { return } rows, err := pool.Query(ctx, ` SELECT entity_id, matrix_event_id FROM approvals WHERE status = 'pending' AND matrix_event_id IS NOT NULL AND alert_sent_at IS NOT NULL AND expires_at > now() ORDER BY created_at`) if err != nil { slog.Error("notifier: query approvals for reactions", "error", err) return } defer rows.Close() for rows.Next() { var approvalID uuid.UUID var matrixEventID string if err := rows.Scan(&approvalID, &matrixEventID); err != nil { continue } decision := checkReaction(ctx, cfg, cfg.MatrixRoomID, matrixEventID) if decision == "" { continue } slog.Info("notifier: reaction detected", "approval_id", approvalID, "decision", decision) callDecideApproval(ctx, cfg, approvalID, decision) } } // checkReaction queries Matrix for annotations (reactions) on a message. func checkReaction(ctx context.Context, cfg config.Config, roomID, eventID string) string { url := fmt.Sprintf("%s/_matrix/client/v3/rooms/%s/relations/%s/m.annotation", cfg.MatrixHomeserver, roomID, eventID) req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { return "" } req.Header.Set("Authorization", "Bearer "+cfg.MatrixToken) resp, err := (&http.Client{Timeout: 10 * time.Second}).Do(req) if err != nil { slog.Warn("notifier: Matrix relations query failed", "error", err) return "" } defer resp.Body.Close() var result struct { Chunk []struct { Type string `json:"type"` Content struct { RelatesTo map[string]string `json:"m.relates_to"` } `json:"content"` } `json:"chunk"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return "" } for _, ev := range result.Chunk { if ev.Type != "m.reaction" { continue } key := ev.Content.RelatesTo["key"] switch { case strings.Contains(key, "\u2705"), strings.Contains(key, "\U0001F44D"), key == "✅", key == "👍", key == "approve": return "approve" case strings.Contains(key, "\u274C"), strings.Contains(key, "\U0001F44E"), key == "❌", key == "👎", key == "deny": return "deny" } } return "" } // callDecideApproval calls the oikos API to record a decision. func callDecideApproval(ctx context.Context, cfg config.Config, approvalID uuid.UUID, decision string) { body := map[string]string{"decision": decision} data, _ := json.Marshal(body) url := fmt.Sprintf("http://api:8090/api/v1/approvals/%s/decision", approvalID) req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(data)) if err != nil { slog.Error("notifier: build decide request", "error", err) return } req.Header.Set("Content-Type", "application/json") if cfg.APIToken != "" { req.Header.Set("Authorization", "Bearer "+cfg.APIToken) } resp, err := (&http.Client{Timeout: 10 * time.Second}).Do(req) if err != nil { slog.Error("notifier: decide API call", "error", err) return } resp.Body.Close() slog.Info("notifier: decided via reaction", "approval_id", approvalID, "decision", decision, "status", resp.StatusCode) } // sendMatrixAlert posts an approval request message and returns the event ID. func sendMatrixAlert(ctx context.Context, cfg config.Config, approvalID uuid.UUID, action, riskClass string) (string, error) { body := fmt.Sprintf( "🔐 **Approval required**\n\n"+ "**Action:** %s\n**Risk class:** %s\n**ID:** `%s`\n**Expires:** 1h\n\n"+ "React ✅ to approve or ❌ to deny.", action, riskClass, approvalID, ) msg := map[string]any{"msgtype": "m.text", "body": body} data, err := json.Marshal(msg) if err != nil { return "", fmt.Errorf("marshal: %w", err) } txnID := fmt.Sprintf("approval-%s-%d", approvalID, time.Now().UnixNano()) url := fmt.Sprintf("%s/_matrix/client/v3/rooms/%s/send/m.room.message/%s", cfg.MatrixHomeserver, cfg.MatrixRoomID, txnID) req, err := http.NewRequestWithContext(ctx, http.MethodPut, url, bytes.NewReader(data)) if err != nil { return "", fmt.Errorf("build request: %w", err) } req.Header.Set("Content-Type", "application/json") req.Header.Set("Authorization", "Bearer "+cfg.MatrixToken) resp, err := (&http.Client{Timeout: 10 * time.Second}).Do(req) if err != nil { return "", fmt.Errorf("send: %w", err) } defer resp.Body.Close() var mxResp struct { EventID string `json:"event_id"` } json.NewDecoder(resp.Body).Decode(&mxResp) if mxResp.EventID == "" { return "", fmt.Errorf("no event_id (status %d)", resp.StatusCode) } slog.Info("notifier: alert sent", "approval_id", approvalID, "event_id", mxResp.EventID) return mxResp.EventID, nil } // generateApprovalToken creates a single-use HMAC token. func generateApprovalToken(approvalID uuid.UUID, secret string) string { if secret == "" { secret = "dev-secret-do-not-use-in-prod" } mac := hmac.New(sha256.New, []byte(secret)) mac.Write([]byte(approvalID.String())) mac.Write([]byte(fmt.Sprintf("%d", time.Now().UnixNano()))) return hex.EncodeToString(mac.Sum(nil)) } func hashToken(token string) string { h := sha256.Sum256([]byte(token)) return hex.EncodeToString(h[:]) }