TDvorak/Primora
1
0 Fork 0
mirror of https://github.com/Dvorinka/Primora.git synced 2026-10-28 15:51:23 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: phase 3 observability — IMS telemetry merge

Browse Source 
Port IMS telemetry into Primora conventions:

- core.components/core.events (migration 00004 + sqlc) scoped to projects
- POST /api/v1/ingest — single or <=500-event batches via project prm_
  keys (X-Primora-Key alias), auto-registering components, permissive CORS
- Issues: fingerprint-grouped errors with count/severity/first+last seen
- Reads: events w/ filters + before-cursor, components, stats (1h/24h/7d/30d
  bucketed series + component health + totals + metric names), metric
  series (avg/p50/p95/max)
- GET .../telemetry/stream — project-scoped SSE through an in-process hub
  (buffered subs, drop-on-slow, 25s keepalives); ?api_key=/?token= auth
  for EventSource
- Telemetry page: health/issues/events/metrics/integrate tabs, live badge,
  390px-clean; demo stubs
- @primora/client SDK (createPrimoraClient: capture*/heartbeat/installAuto,
  queued keepalive flushes)
- MCP: primora_list_issues/list_events/telemetry_stats/metric_series
- IMS README deprecation note pointing here

Also fixes workspace pins to 0.3.0 — npm ci in the frontend image was
resolving the unpublished 0.2.0 spec from the registry.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Tomas DvorakandDevin committed 2026-09-16 16:21:03 +02:00
1 parent 6507eee606
commit 8cfddf9184
60 files changed
+3694 -30
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

apps/backend/internal/services/telemetry.go
+459
View File
@@ -0,0 +1,459 @@
package services
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
db "github.com/tdvorak/primora/apps/backend/internal/database/db"
"github.com/tdvorak/primora/apps/backend/internal/models"
)
var telemetryEventTypes = map[string]bool{
"error": true, "metric": true, "log": true, "heartbeat": true, "event": true,
}
var componentKinds = map[string]bool{
"frontend": true, "backend": true, "database": true,
"android": true, "desktop": true, "web": true, "other": true,
}
// IngestEventInput is one event in an ingest batch.
type IngestEventInput struct {
Type string `json:"type"`
Component string `json:"component"`
Kind string `json:"kind"`
Severity string `json:"severity"`
Message string `json:"message"`
Payload json.RawMessage `json:"payload"`
Fingerprint string `json:"fingerprint"`
TS *time.Time `json:"ts"`
}
// TelemetryEvent is the client-facing event shape.
type TelemetryEvent struct {
ID int64 `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
ComponentID *uuid.UUID `json:"component_id"`
ComponentName string `json:"component_name,omitempty"`
Type string `json:"type"`
Severity string `json:"severity"`
Message string `json:"message"`
Payload json.RawMessage `json:"payload"`
Fingerprint string `json:"fingerprint"`
TS time.Time `json:"ts"`
}
type TelemetryIssue struct {
Fingerprint string `json:"fingerprint"`
Message string `json:"message"`
ComponentID *uuid.UUID `json:"component_id"`
ComponentName string `json:"component_name,omitempty"`
Count int64 `json:"count"`
FirstSeen time.Time `json:"first_seen"`
LastSeen time.Time `json:"last_seen"`
Severity string `json:"severity"`
}
type ComponentSummary struct {
ID uuid.UUID `json:"id"`
Name string `json:"name"`
Kind string `json:"kind"`
LastSeenAt *time.Time `json:"last_seen_at"`
LastStatus string `json:"last_status"`
Errors24h int64 `json:"errors_24h"`
Events24h int64 `json:"events_24h"`
Meta map[string]any `json:"meta"`
}
func fingerprintOf(componentID uuid.UUID, message string) string {
line := strings.SplitN(message, "\n", 2)[0]
sum := sha256.Sum256([]byte(componentID.String() + "|" + line))
return hex.EncodeToString(sum[:])[:16]
}
// IngestEvents stores a batch of telemetry events under the ingest key's
// project. API-key actors only — the key's project scope is the write scope.
func (s *PlatformService) IngestEvents(ctx context.Context, actor *models.Actor, events []IngestEventInput) (int, error) {
if actor == nil || actor.ProjectID == nil || actor.APIKeyID == nil {
return 0, errors.New("ingest requires a project API key")
}
projectID := *actor.ProjectID
var inserted []TelemetryEvent
err := s.repo.WithTx(ctx, func(q *db.Queries) error {
compCache := map[string]*uuid.UUID{}
for _, ev := range events {
if !telemetryEventTypes[ev.Type] {
continue
}
var compID *uuid.UUID
if ev.Component != "" {
cached, hit := compCache[ev.Component]
if !hit {
kind := ev.Kind
if !componentKinds[kind] {
kind = "other"
}
row, err := q.UpsertComponent(ctx, db.UpsertComponentParams{
ProjectID: projectID,
Name: ev.Component,
Kind: kind,
})
if err != nil {
continue
}
id := row.ID
cached = &id
compCache[ev.Component] = cached
}
compID = cached
}
sev := ev.Severity
if sev == "" {
sev = "info"
if ev.Type == "error" {
sev = "error"
}
}
fp := ev.Fingerprint
if ev.Type == "error" && fp == "" && compID != nil {
fp = fingerprintOf(*compID, ev.Message)
}
ts := time.Now()
if ev.TS != nil {
ts = *ev.TS
}
payload := ev.Payload
if len(payload) == 0 {
payload = json.RawMessage(`{}`)
}
var cid pgtype.UUID
if compID != nil {
cid = pgtype.UUID{Bytes: *compID, Valid: true}
}
row, err := q.InsertEvent(ctx, db.InsertEventParams{
ProjectID: projectID,
ComponentID: cid,
Type: ev.Type,
Severity: sev,
Message: ev.Message,
Payload: payload,
Fingerprint: fp,
Ts: pgtype.Timestamptz{Time: ts, Valid: true},
})
if err != nil {
continue
}
out := TelemetryEvent{
ID: row.ID,
ProjectID: row.ProjectID,
Type: row.Type,
Severity: row.Severity,
Message: row.Message,
Payload: row.Payload,
Fingerprint: row.Fingerprint,
TS: row.Ts.Time,
}
if compID != nil {
out.ComponentID = compID
out.ComponentName = ev.Component
}
inserted = append(inserted, out)
}
return nil
})
if err != nil {
return 0, err
}
_ = s.repo.Queries().TouchAPIKey(ctx, *actor.APIKeyID)
for _, ev := range inserted {
s.hub.Broadcast(ev.ProjectID.String(), ev)
}
return len(inserted), nil
}
// SubscribeTelemetry validates read access and returns an event channel.
func (s *PlatformService) SubscribeTelemetry(ctx context.Context, actor *models.Actor, projectID uuid.UUID) (chan []byte, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
return s.hub.Subscribe(projectID.String()), nil
}
func (s *PlatformService) UnsubscribeTelemetry(projectID uuid.UUID, ch chan []byte) {
s.hub.Unsubscribe(projectID.String(), ch)
}
type EventListFilters struct {
Type string
Component string
Fingerprint string
Before *int64
Limit int
}
func (s *PlatformService) ListEvents(ctx context.Context, actor *models.Actor, projectID uuid.UUID, f EventListFilters) ([]TelemetryEvent, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
if f.Limit <= 0 || f.Limit > 500 {
f.Limit = 100
}
rows, err := s.repo.Queries().ListEvents(ctx, db.ListEventsParams{
ProjectID: projectID,
Type: strPtr(f.Type),
Component: strPtr(f.Component),
Fingerprint: strPtr(f.Fingerprint),
Before: f.Before,
Limit: int32(f.Limit),
})
if err != nil {
return nil, err
}
out := make([]TelemetryEvent, 0, len(rows))
for _, r := range rows {
out = append(out, toTelemetryEvent(r))
}
return out, nil
}
func (s *PlatformService) ListIssues(ctx context.Context, actor *models.Actor, projectID uuid.UUID, days int) ([]TelemetryIssue, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
if days <= 0 || days > 365 {
days = 30
}
rows, err := s.repo.Queries().ListErrorGroups(ctx, db.ListErrorGroupsParams{
ProjectID: projectID,
Days: strPtr(fmt.Sprint(days)),
})
if err != nil {
return nil, err
}
out := make([]TelemetryIssue, 0, len(rows))
for _, r := range rows {
issue := TelemetryIssue{
Fingerprint: r.Fingerprint,
Message: r.Message,
Count: r.Count,
Severity: r.Severity,
}
if r.FirstSeen.Valid {
issue.FirstSeen = r.FirstSeen.Time
}
if r.LastSeen.Valid {
issue.LastSeen = r.LastSeen.Time
}
if id, err := uuid.Parse(r.ComponentID); err == nil {
issue.ComponentID = &id
}
issue.ComponentName = r.ComponentName
out = append(out, issue)
}
return out, nil
}
func windowSeconds(w string) int64 {
switch w {
case "1h":
return 3600
case "7d":
return 7 * 86400
case "30d":
return 30 * 86400
default:
return 86400
}
}
func bucketSeconds(window int64) int64 {
switch {
case window <= 3600:
return 60
case window <= 86400:
return 900
case window <= 7*86400:
return 3600
default:
return 86400
}
}
type EventSeriesBucket struct {
TS time.Time `json:"ts"`
Counts map[string]int64 `json:"counts"`
}
type MetricPoint struct {
TS time.Time `json:"ts"`
Avg *float64 `json:"avg"`
P50 *float64 `json:"p50"`
P95 *float64 `json:"p95"`
Max *float64 `json:"max"`
Count int64 `json:"count"`
}
type TelemetryStatsResult struct {
Window int64 `json:"window"`
BucketSec int64 `json:"bucket_sec"`
Series []EventSeriesBucket `json:"series"`
Components []ComponentSummary `json:"components"`
Errors int64 `json:"errors"`
Events int64 `json:"events"`
Metrics int64 `json:"metrics"`
MetricNames []string `json:"metric_names"`
}
func (s *PlatformService) TelemetryStats(ctx context.Context, actor *models.Actor, projectID uuid.UUID, window string) (*TelemetryStatsResult, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
win := windowSeconds(window)
bucket := bucketSeconds(win)
seriesRows, err := s.repo.Queries().EventTypeSeries(ctx, db.EventTypeSeriesParams{
ProjectID: projectID,
BucketSec: bucket,
WindowSec: win,
})
if err != nil {
return nil, err
}
series := make([]EventSeriesBucket, 0, len(seriesRows))
for _, r := range seriesRows {
n := len(series)
if n == 0 || !series[n-1].TS.Equal(r.Bucket.Time) {
series = append(series, EventSeriesBucket{TS: r.Bucket.Time, Counts: map[string]int64{}})
n++
}
series[n-1].Counts[r.Type] = r.Count
}
compRows, err := s.repo.Queries().ComponentHealthRows(ctx, projectID)
if err != nil {
return nil, err
}
components := make([]ComponentSummary, 0, len(compRows))
for _, r := range compRows {
c := ComponentSummary{ID: r.ID, Name: r.Name, Kind: r.Kind, Errors24h: r.Errors24h, Events24h: r.Events24h}
if r.LastSeenAt.Valid {
t := r.LastSeenAt.Time
c.LastSeenAt = &t
}
c.LastStatus = r.LastStatus
if len(r.Meta) > 0 {
_ = json.Unmarshal(r.Meta, &c.Meta)
}
components = append(components, c)
}
totals, err := s.repo.Queries().EventTotals(ctx, db.EventTotalsParams{ProjectID: projectID, WindowSec: win})
if err != nil {
return nil, err
}
names, err := s.repo.Queries().MetricNames(ctx, projectID)
if err != nil {
return nil, err
}
if names == nil {
names = []string{}
}
return &TelemetryStatsResult{
Window: win,
BucketSec: bucket,
Series: series,
Components: components,
Errors: totals.Errors,
Events: totals.Events,
Metrics: totals.Metrics,
MetricNames: names,
}, nil
}
func (s *PlatformService) MetricSeriesData(ctx context.Context, actor *models.Actor, projectID uuid.UUID, name, window string) ([]MetricPoint, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
if name == "" {
return nil, errors.New("name required")
}
win := windowSeconds(window)
rows, err := s.repo.Queries().MetricSeries(ctx, db.MetricSeriesParams{
ProjectID: projectID,
Name: name,
BucketSec: bucketSeconds(win),
WindowSec: win,
})
if err != nil {
return nil, err
}
out := make([]MetricPoint, 0, len(rows))
for _, r := range rows {
out = append(out, MetricPoint{TS: r.Bucket.Time, Avg: &r.Avg, P50: &r.P50, P95: &r.P95, Max: &r.Max, Count: r.Count})
}
return out, nil
}
func (s *PlatformService) ListComponents(ctx context.Context, actor *models.Actor, projectID uuid.UUID) ([]ComponentSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
rows, err := s.repo.Queries().ListComponents(ctx, projectID)
if err != nil {
return nil, err
}
out := make([]ComponentSummary, 0, len(rows))
for _, r := range rows {
c := ComponentSummary{ID: r.ID, Name: r.Name, Kind: r.Kind}
if len(r.Meta) > 0 {
_ = json.Unmarshal(r.Meta, &c.Meta)
}
out = append(out, c)
}
return out, nil
}
func (s *PlatformService) DeleteComponent(ctx context.Context, actor *models.Actor, projectID, componentID uuid.UUID) error {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return err
}
return s.repo.Queries().DeleteComponent(ctx, db.DeleteComponentParams{ID: componentID, ProjectID: projectID})
}
func toTelemetryEvent(r db.ListEventsRow) TelemetryEvent {
ev := TelemetryEvent{
ID: r.ID,
ProjectID: r.ProjectID,
Type: r.Type,
Severity: r.Severity,
Message: r.Message,
Payload: r.Payload,
Fingerprint: r.Fingerprint,
TS: r.Ts.Time,
}
if r.ComponentID.Valid {
id := uuid.UUID(r.ComponentID.Bytes)
ev.ComponentID = &id
}
if r.ComponentName != nil {
ev.ComponentName = *r.ComponentName
}
return ev
}
func strPtr(s string) *string {
if s == "" {
return nil
}
return &s
}