diff --git a/lib/api/echo/basicauth.go b/lib/api/echo/basicauth.go index 96a36bc..20499d1 100644 --- a/lib/api/echo/basicauth.go +++ b/lib/api/echo/basicauth.go @@ -5,8 +5,10 @@ import ( "github.com/labstack/echo/v5/middleware" ) -type User string -type Password string +type ( + User string + Password string +) type RestUsers struct { RestUser map[User]Password `long:"http-rest-user" env:"HTTP_REST_USER" description:"Usernames for REST API access."` diff --git a/lib/events/avro/serde.go b/lib/events/avro/serde.go index bf58e74..ebb4343 100644 --- a/lib/events/avro/serde.go +++ b/lib/events/avro/serde.go @@ -11,8 +11,10 @@ import ( confluent "github.com/confluentinc/confluent-kafka-go/v2/schemaregistry" ) -var ErrPayloadToShort = errors.New("payload too short") -var ErrInvalidMagicByte = errors.New("invalid magic byte value") +var ( + ErrPayloadToShort = errors.New("payload too short") + ErrInvalidMagicByte = errors.New("invalid magic byte value") +) type Event interface { // Schema returns the avro schema of this event diff --git a/lib/events/event-topics.go b/lib/events/event-topics.go index 96aa1a3..ab14213 100644 --- a/lib/events/event-topics.go +++ b/lib/events/event-topics.go @@ -10,8 +10,10 @@ import ( // TopicsFunc builds an EventTopics instance for the given kafka replication factor. type TopicsFunc func(replicationFactor int16) EventTopics -type Topics = startup_kafka.Topics -type Topic = startup_kafka.Topic +type ( + Topics = startup_kafka.Topics + Topic = startup_kafka.Topic +) // EventTopics contains a mapping from event struct type to a kafka topic. // This map must contain all event types that are going to be send. If this misses an event, diff --git a/lib/history/render.go b/lib/history/render.go new file mode 100644 index 0000000..630e8ca --- /dev/null +++ b/lib/history/render.go @@ -0,0 +1,127 @@ +package history + +import ( + "bytes" + "context" + "embed" + "encoding/json" + "fmt" + "html/template" + "io" + "sort" + "time" + + "github.com/flachnetz/startup/v2/lib/ql" +) + +//go:embed templates/history.gohtml +var templateFS embed.FS + +//go:embed templates/overview.gohtml +var overviewFS embed.FS + +var pageTemplate = template.Must(template.New("history.gohtml"). + Funcs(template.FuncMap{ + "formatTime": func(t time.Time) string { return t.Format("2006-01-02 15:04:05.000") }, + }). + ParseFS(templateFS, "templates/history.gohtml")) + +var overviewTemplate = template.Must(template.New("overview.gohtml"). + ParseFS(overviewFS, "templates/overview.gohtml")) + +// OverviewModel is the template model for a clickable list page linking to +// per-item detail pages. +type OverviewModel struct { + Title string + Headers []string + Rows []OverviewRow +} + +// OverviewRow is one list entry; Cells aligns with OverviewModel.Headers and +// Link is the detail-page URL the row navigates to. +type OverviewRow struct { + Link string + Cells []string +} + +// RenderOverview writes a standalone clickable table; each row links to Link. +func RenderOverview(w io.Writer, title string, headers []string, rows []OverviewRow) error { + if err := overviewTemplate.Execute(w, OverviewModel{Title: title, Headers: headers, Rows: rows}); err != nil { + return fmt.Errorf("render overview: %w", err) + } + return nil +} + +// RecordView wraps a Record with its pretty-printed JSON payload for rendering. +type RecordView struct { + Record + // JSON is the indented payload. + JSON string + // ShowSeparator is true when this record starts a new RequestTraceId group. + ShowSeparator bool +} + +// PageModel is the template model for the generic history page. +type PageModel struct { + Title string + GroupId string + ErrorMessage string + Summary []SummaryItem + Records []RecordView +} + +// SummaryItem is one label/value row shown above the ledger, describing the +// current state of the tracked object. Ordered slice (not a map) so the page +// renders stably. +type SummaryItem struct { + Label string + Value string +} + +// RenderPage writes a standalone HTML history page for groupId to w. Records are +// loaded in a new read transaction, sorted by Timestamp (Service.Records is +// unordered), and an
separates consecutive RequestTraceId groups. +// +// ponytail: payload is rendered as pretty JSON only; add a key/value table when +// an item needs structured display. +func (h *Service) RenderPage(ctx context.Context, w io.Writer, groupId GroupId, title string) error { + return h.RenderPageSummary(ctx, w, groupId, title, nil) +} + +// RenderPageSummary is RenderPage with an extra current-state summary rendered +// above the ledger. +func (h *Service) RenderPageSummary(ctx context.Context, w io.Writer, groupId GroupId, title string, summary []SummaryItem) error { + records, err := ql.InNewTransactionWithResult(ctx, h.txStarter, func(ctx ql.TxContext) ([]Record, error) { + return h.Records(ctx, groupId) + }) + if err != nil { + return fmt.Errorf("load records: %w", err) + } + + sort.SliceStable(records, func(i, j int) bool { + return records[i].Timestamp.Before(records[j].Timestamp) + }) + + views := make([]RecordView, len(records)) + for i, rec := range records { + var pretty bytes.Buffer + if err := json.Indent(&pretty, rec.Payload, "", " "); err != nil { + // keep the raw payload if it is not valid JSON. + pretty.Reset() + pretty.Write(rec.Payload) + } + + views[i] = RecordView{ + Record: rec, + JSON: pretty.String(), + ShowSeparator: i > 0 && rec.RequestTraceId.String() != records[i-1].RequestTraceId.String(), + } + } + + return pageTemplate.Execute(w, PageModel{ + Title: title, + GroupId: groupId.String(), + Summary: summary, + Records: views, + }) +} diff --git a/lib/history/singleton.go b/lib/history/singleton.go index a5c0c0c..d2ebceb 100644 --- a/lib/history/singleton.go +++ b/lib/history/singleton.go @@ -2,7 +2,9 @@ package history import ( "context" + "errors" "fmt" + "io" "log/slog" "github.com/flachnetz/startup/v2/lib/events" @@ -34,7 +36,6 @@ func InitializeGlobal(ctx context.Context, opts Options) error { err := ql.InNewTransaction(ctx, opts.DB, func(ctx ql.TxContext) error { return CreateTable(ctx, opts.HistoryTable) }) - if err != nil { return fmt.Errorf("create history table: %w", err) } @@ -65,3 +66,22 @@ func Track[T ~string](ctx context.Context, groupId T, item Item) { instance.Track(ctx, GroupId(groupId), item) } + +// RenderPage uses the global history singleton to render the history page for +// groupId. You need to initialize it using InitializeGlobal first. +func RenderPage[T ~string](ctx context.Context, w io.Writer, groupId T, title string) error { + if instance == nil { + return errors.New("history: global instance not initialized") + } + + return instance.RenderPage(ctx, w, GroupId(groupId), title) +} + +// RenderPageSummary is RenderPage with a current-state summary above the ledger. +func RenderPageSummary[T ~string](ctx context.Context, w io.Writer, groupId T, title string, summary []SummaryItem) error { + if instance == nil { + return errors.New("history: global instance not initialized") + } + + return instance.RenderPageSummary(ctx, w, GroupId(groupId), title, summary) +} diff --git a/lib/history/templates/history.gohtml b/lib/history/templates/history.gohtml new file mode 100644 index 0000000..2f5c584 --- /dev/null +++ b/lib/history/templates/history.gohtml @@ -0,0 +1,55 @@ + + + + + + {{ .Title }} + + + + + +

{{ .Title }}

+

GroupId: {{ .GroupId }}

+{{ if .Summary }} +
+ {{ range .Summary }}
{{ .Label }}{{ .Value }}
{{ end }} +
+{{ end }} +{{ if .ErrorMessage }}

{{ .ErrorMessage }}

{{ end }} +
+ {{ range .Records }} + {{ if .ShowSeparator }}
{{ end }} +
+

{{ .Step }}

+

{{ .Description }}

+

{{ formatTime .Timestamp }} · TraceId: {{ .RequestTraceId }} · EventSender: {{ .EventSender }}{{ if .EventSenderVersion }} ({{ .EventSenderVersion }}){{ end }}

+
+ payload + 
{{ .JSON }}
+
+
+ {{ else }} +

No history records.

+ {{ end }} +
+ + + + diff --git a/lib/history/templates/overview.gohtml b/lib/history/templates/overview.gohtml new file mode 100644 index 0000000..786436f --- /dev/null +++ b/lib/history/templates/overview.gohtml @@ -0,0 +1,29 @@ + + + + + {{ .Title }} + + + +

{{ .Title }}

+ + {{ range .Headers }}{{ end }} + {{ range .Rows }} + + {{ range .Cells }}{{ end }} + + {{ else }} + + {{ end }} +
{{ . }}
{{ . }}
No records.
+ + diff --git a/lib/kconsumer/consumer.go b/lib/kconsumer/consumer.go index 41909bd..7fff71e 100644 --- a/lib/kconsumer/consumer.go +++ b/lib/kconsumer/consumer.go @@ -142,7 +142,6 @@ func runWorker(ctx context.Context, w *partitionWorker, handle HandleMessage) { return }) - if err != nil { log.Error("Giving up on message", slog.Int64("offset", int64(offset))) w.errCh <- err diff --git a/startup_kafka/options.go b/startup_kafka/options.go index e841bfe..57f074e 100644 --- a/startup_kafka/options.go +++ b/startup_kafka/options.go @@ -67,7 +67,7 @@ func (opts *KafkaOptions) NewConsumer(overrideConfig kafka.ConfigMap) *kafka.Con // Resolve the special RANDOM group into a unique group id once, so repeated // calls keep using the same generated group. - var consumerGroup = opts.KafkaConsumerGroup + consumerGroup := opts.KafkaConsumerGroup if consumerGroup == "RANDOM" { consumerGroup = fmt.Sprintf("golang-%d", time.Now().UnixNano()) }