feat: split workbench runtime read service
This commit is contained in:
lyon
@@ -0,0 +1,725 @@
|
||||
package workbenchruntime
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
_ "github.com/jackc/pgx/v4/stdlib"
|
||||
)
|
||||
|
||||
const serviceID = "hwlab-workbench-runtime"
|
||||
|
||||
type Config struct {
|
||||
Addr string
|
||||
DatabaseURL string
|
||||
QueryTimeout time.Duration
|
||||
ReadinessDelay time.Duration
|
||||
PoolMax int
|
||||
}
|
||||
|
||||
type Server struct {
|
||||
config Config
|
||||
db *sql.DB
|
||||
mux *http.ServeMux
|
||||
}
|
||||
|
||||
type factQuery struct {
|
||||
Families []string `json:"families"`
|
||||
FactFamilies []string `json:"factFamilies"`
|
||||
Limit int `json:"limit"`
|
||||
AfterProjectedSeq int64 `json:"afterProjectedSeq"`
|
||||
Order string `json:"order"`
|
||||
SessionsOrder string `json:"sessionsOrder"`
|
||||
MessagesOrder string `json:"messagesOrder"`
|
||||
PartsOrder string `json:"partsOrder"`
|
||||
TurnsOrder string `json:"turnsOrder"`
|
||||
CheckpointsOrder string `json:"checkpointsOrder"`
|
||||
SessionProjection string `json:"sessionProjection"`
|
||||
SessionID string `json:"sessionId"`
|
||||
SessionIDs []string `json:"sessionIds"`
|
||||
TraceID string `json:"traceId"`
|
||||
TraceIDs []string `json:"traceIds"`
|
||||
MessageID string `json:"messageId"`
|
||||
MessageIDs []string `json:"messageIds"`
|
||||
TurnID string `json:"turnId"`
|
||||
TurnIDs []string `json:"turnIds"`
|
||||
Status string `json:"status"`
|
||||
Statuses []string `json:"statuses"`
|
||||
OwnerUserID string `json:"ownerUserId"`
|
||||
OwnerUserIDs []string `json:"ownerUserIds"`
|
||||
ConversationID string `json:"conversationId"`
|
||||
ConversationIDs []string `json:"conversationIds"`
|
||||
EventType string `json:"eventType"`
|
||||
EventTypes []string `json:"eventTypes"`
|
||||
ProjectionStatus string `json:"projectionStatus"`
|
||||
ProjectionStatuses []string `json:"projectionStatuses"`
|
||||
ProjectionHealth string `json:"projectionHealth"`
|
||||
ProjectionHealths []string `json:"projectionHealths"`
|
||||
RunID string `json:"runId"`
|
||||
RunIDs []string `json:"runIds"`
|
||||
CommandID string `json:"commandId"`
|
||||
CommandIDs []string `json:"commandIds"`
|
||||
}
|
||||
|
||||
type traceEventsQuery struct {
|
||||
TraceID string `json:"traceId"`
|
||||
SessionID string `json:"sessionId"`
|
||||
WorkerSessionID string `json:"workerSessionId"`
|
||||
Level string `json:"level"`
|
||||
Limit int `json:"limit"`
|
||||
}
|
||||
|
||||
type outboxQuery struct {
|
||||
AfterSeq int64 `json:"afterSeq"`
|
||||
Limit int `json:"limit"`
|
||||
TraceID string `json:"traceId"`
|
||||
SessionID string `json:"sessionId"`
|
||||
}
|
||||
|
||||
type fieldColumn struct {
|
||||
field string
|
||||
values []string
|
||||
column string
|
||||
}
|
||||
|
||||
func Run(ctx context.Context) error {
|
||||
config := configFromEnv()
|
||||
if strings.TrimSpace(config.DatabaseURL) == "" {
|
||||
return errors.New("HWLAB_WORKBENCH_RUNTIME_DB_URL or HWLAB_CLOUD_DB_URL is required")
|
||||
}
|
||||
db, err := sql.Open("pgx", config.DatabaseURL)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer db.Close()
|
||||
if config.PoolMax > 0 {
|
||||
db.SetMaxOpenConns(config.PoolMax)
|
||||
db.SetMaxIdleConns(config.PoolMax)
|
||||
}
|
||||
server := NewServer(config, db)
|
||||
httpServer := &http.Server{Addr: config.Addr, Handler: server.mux, ReadHeaderTimeout: 5 * time.Second}
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
log.Printf("%s listening addr=%s", serviceID, config.Addr)
|
||||
errCh <- httpServer.ListenAndServe()
|
||||
}()
|
||||
stopCh := make(chan os.Signal, 1)
|
||||
signal.Notify(stopCh, syscall.SIGINT, syscall.SIGTERM)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-stopCh:
|
||||
case err := <-errCh:
|
||||
if err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
return httpServer.Shutdown(shutdownCtx)
|
||||
}
|
||||
|
||||
func NewServer(config Config, db *sql.DB) *Server {
|
||||
server := &Server{config: config, db: db, mux: http.NewServeMux()}
|
||||
server.routes()
|
||||
return server
|
||||
}
|
||||
|
||||
func (s *Server) routes() {
|
||||
s.mux.HandleFunc("/health/live", s.handleLive)
|
||||
s.mux.HandleFunc("/health/ready", s.handleReady)
|
||||
s.mux.HandleFunc("/v1/workbench-runtime/facts", s.handleFacts)
|
||||
s.mux.HandleFunc("/v1/workbench-runtime/trace-events", s.handleTraceEvents)
|
||||
s.mux.HandleFunc("/v1/workbench-runtime/projection-outbox", s.handleProjectionOutbox)
|
||||
}
|
||||
|
||||
func (s *Server) handleLive(w http.ResponseWriter, r *http.Request) {
|
||||
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "serviceId": serviceID, "status": "live", "valuesRedacted": true})
|
||||
}
|
||||
|
||||
func (s *Server) handleReady(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), s.config.ReadinessDelay)
|
||||
defer cancel()
|
||||
if err := s.db.PingContext(ctx); err != nil {
|
||||
writeAPIError(w, http.StatusServiceUnavailable, "workbench_runtime_db_unavailable", "workbench runtime database is unavailable")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{"ok": true, "serviceId": serviceID, "status": "ready", "valuesRedacted": true})
|
||||
}
|
||||
|
||||
func (s *Server) handleFacts(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
methodNotAllowed(w, "POST")
|
||||
return
|
||||
}
|
||||
var query factQuery
|
||||
if !decodeJSON(w, r, &query) {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout)
|
||||
defer cancel()
|
||||
facts, err := s.queryFacts(ctx, query)
|
||||
if err != nil {
|
||||
writeQueryError(w, err)
|
||||
return
|
||||
}
|
||||
count := 0
|
||||
for _, rows := range facts {
|
||||
count += len(rows)
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"contractVersion": "workbench-runtime-facts-v1",
|
||||
"facts": facts,
|
||||
"count": count,
|
||||
"persistence": s.persistenceSummary(),
|
||||
"valuesRedacted": true,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) handleTraceEvents(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
methodNotAllowed(w, "POST")
|
||||
return
|
||||
}
|
||||
var query traceEventsQuery
|
||||
if !decodeJSON(w, r, &query) {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout)
|
||||
defer cancel()
|
||||
events, err := s.queryAgentTraceEvents(ctx, query)
|
||||
if err != nil {
|
||||
writeQueryError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"contractVersion": "workbench-runtime-trace-events-v1",
|
||||
"events": events,
|
||||
"count": len(events),
|
||||
"persistence": s.persistenceSummary(),
|
||||
"valuesRedacted": true,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) handleProjectionOutbox(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
methodNotAllowed(w, "POST")
|
||||
return
|
||||
}
|
||||
var query outboxQuery
|
||||
if !decodeJSON(w, r, &query) {
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(r.Context(), s.config.QueryTimeout)
|
||||
defer cancel()
|
||||
rows, err := s.readProjectionOutbox(ctx, query)
|
||||
if err != nil {
|
||||
writeQueryError(w, err)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, map[string]any{
|
||||
"contractVersion": "workbench-runtime-projection-outbox-v1",
|
||||
"rows": rows,
|
||||
"count": len(rows),
|
||||
"persistence": s.persistenceSummary(),
|
||||
"valuesRedacted": true,
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Server) queryFacts(ctx context.Context, query factQuery) (map[string][]any, error) {
|
||||
families := factFamilySet(query)
|
||||
facts := emptyFacts()
|
||||
var err error
|
||||
if families["sessions"] {
|
||||
facts["sessions"], err = s.queryFactRows(ctx, "workbench_sessions", "session_json", query, []fieldColumn{
|
||||
{"sessionId", query.SessionIDs, "session_id"},
|
||||
{"traceId", query.TraceIDs, "last_trace_id"},
|
||||
{"ownerUserId", query.OwnerUserIDs, "owner_user_id"},
|
||||
{"conversationId", query.ConversationIDs, "conversation_id"},
|
||||
{"status", query.Statuses, "status"},
|
||||
})
|
||||
if err != nil { return nil, err }
|
||||
}
|
||||
if families["messages"] {
|
||||
facts["messages"], err = s.queryFactRows(ctx, "workbench_messages", "message_json", query, []fieldColumn{
|
||||
{"messageId", query.MessageIDs, "message_id"},
|
||||
{"sessionId", query.SessionIDs, "session_id"},
|
||||
{"traceId", query.TraceIDs, "trace_id"},
|
||||
{"turnId", query.TurnIDs, "turn_id"},
|
||||
{"status", query.Statuses, "status"},
|
||||
})
|
||||
if err != nil { return nil, err }
|
||||
}
|
||||
if families["parts"] {
|
||||
facts["parts"], err = s.queryFactRows(ctx, "workbench_parts", "part_json", query, []fieldColumn{
|
||||
{"messageId", query.MessageIDs, "message_id"},
|
||||
{"sessionId", query.SessionIDs, "session_id"},
|
||||
{"traceId", query.TraceIDs, "trace_id"},
|
||||
{"turnId", query.TurnIDs, "turn_id"},
|
||||
{"status", query.Statuses, "status"},
|
||||
})
|
||||
if err != nil { return nil, err }
|
||||
}
|
||||
if families["turns"] {
|
||||
facts["turns"], err = s.queryFactRows(ctx, "workbench_turns", "turn_json", query, []fieldColumn{
|
||||
{"turnId", query.TurnIDs, "turn_id"},
|
||||
{"sessionId", query.SessionIDs, "session_id"},
|
||||
{"traceId", query.TraceIDs, "trace_id"},
|
||||
{"messageId", query.MessageIDs, "message_id"},
|
||||
{"status", query.Statuses, "status"},
|
||||
})
|
||||
if err != nil { return nil, err }
|
||||
}
|
||||
if families["traceEvents"] {
|
||||
facts["traceEvents"], err = s.queryFactRows(ctx, "workbench_trace_events", "event_json", query, []fieldColumn{
|
||||
{"traceId", query.TraceIDs, "trace_id"},
|
||||
{"sessionId", query.SessionIDs, "session_id"},
|
||||
{"turnId", query.TurnIDs, "turn_id"},
|
||||
{"messageId", query.MessageIDs, "message_id"},
|
||||
{"eventType", query.EventTypes, "event_type"},
|
||||
})
|
||||
if err != nil { return nil, err }
|
||||
}
|
||||
if families["checkpoints"] {
|
||||
facts["checkpoints"], err = s.queryFactRows(ctx, "workbench_projection_checkpoints", "checkpoint_json", query, []fieldColumn{
|
||||
{"traceId", query.TraceIDs, "trace_id"},
|
||||
{"sessionId", query.SessionIDs, "session_id"},
|
||||
{"turnId", query.TurnIDs, "turn_id"},
|
||||
{"runId", query.RunIDs, "run_id"},
|
||||
{"commandId", query.CommandIDs, "command_id"},
|
||||
{"projectionStatus", query.ProjectionStatuses, "projection_status"},
|
||||
{"projectionHealth", query.ProjectionHealths, "projection_health"},
|
||||
})
|
||||
if err != nil { return nil, err }
|
||||
}
|
||||
return facts, nil
|
||||
}
|
||||
|
||||
func (s *Server) queryFactRows(ctx context.Context, table string, jsonColumn string, query factQuery, fields []fieldColumn) ([]any, error) {
|
||||
clauses, args := factClauses(query, fields)
|
||||
limit := boundedLimit(query.Limit, 0, 1000)
|
||||
if table == "workbench_trace_events" {
|
||||
if query.AfterProjectedSeq > 0 {
|
||||
args = append(args, query.AfterProjectedSeq)
|
||||
clauses = append(clauses, fmt.Sprintf("projected_seq > $%d", len(args)))
|
||||
}
|
||||
sqlText := fmt.Sprintf("SELECT %s FROM %s%s ORDER BY projected_seq ASC", jsonColumn, table, whereSQL(clauses))
|
||||
if limit > 0 {
|
||||
args = append(args, limit)
|
||||
sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
|
||||
}
|
||||
return s.queryJSONColumn(ctx, sqlText, args)
|
||||
}
|
||||
orderDirection := "ASC"
|
||||
if factOrder(query, factFamilyForTable(table)) == "updated_desc" {
|
||||
orderDirection = "DESC"
|
||||
}
|
||||
if table == "workbench_sessions" && query.SessionProjection == "summary" {
|
||||
sqlText := fmt.Sprintf("SELECT %s FROM workbench_sessions%s ORDER BY updated_at %s", sessionSummarySelectClause(), whereSQL(clauses), orderDirection)
|
||||
if limit > 0 {
|
||||
args = append(args, limit)
|
||||
sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
|
||||
}
|
||||
return s.querySessionSummaries(ctx, sqlText, args)
|
||||
}
|
||||
sqlText := fmt.Sprintf("SELECT %s FROM %s%s ORDER BY updated_at %s", jsonColumn, table, whereSQL(clauses), orderDirection)
|
||||
if limit > 0 {
|
||||
args = append(args, limit)
|
||||
sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
|
||||
}
|
||||
return s.queryJSONColumn(ctx, sqlText, args)
|
||||
}
|
||||
|
||||
func (s *Server) queryJSONColumn(ctx context.Context, sqlText string, args []any) ([]any, error) {
|
||||
rows, err := s.db.QueryContext(ctx, sqlText, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
result := []any{}
|
||||
for rows.Next() {
|
||||
var raw []byte
|
||||
if err := rows.Scan(&raw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var item any
|
||||
if len(raw) > 0 && json.Unmarshal(raw, &item) == nil && item != nil {
|
||||
result = append(result, item)
|
||||
}
|
||||
}
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Server) querySessionSummaries(ctx context.Context, sqlText string, args []any) ([]any, error) {
|
||||
rows, err := s.db.QueryContext(ctx, sqlText, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
result := []any{}
|
||||
for rows.Next() {
|
||||
var sessionID, ownerUserID, projectID, conversationID, threadID, status, lastTraceID, sourceEventID, providerProfile sql.NullString
|
||||
var projectedSeq, sourceSeq sql.NullInt64
|
||||
var terminal, sealed sql.NullBool
|
||||
var createdAt, updatedAt sql.NullTime
|
||||
if err := rows.Scan(&sessionID, &ownerUserID, &projectID, &conversationID, &threadID, &status, &lastTraceID, &projectedSeq, &sourceSeq, &sourceEventID, &terminal, &sealed, &createdAt, &updatedAt, &providerProfile); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !sessionID.Valid || strings.TrimSpace(sessionID.String) == "" {
|
||||
continue
|
||||
}
|
||||
item := map[string]any{
|
||||
"id": sessionID.String,
|
||||
"sessionId": sessionID.String,
|
||||
"ownerUserId": nullableString(ownerUserID),
|
||||
"projectId": nullableString(projectID),
|
||||
"conversationId": nullableString(conversationID),
|
||||
"threadId": nullableString(threadID),
|
||||
"status": first(nullableStringValue(status), "unknown"),
|
||||
"lastTraceId": nullableString(lastTraceID),
|
||||
"projectedSeq": nullableInt(projectedSeq),
|
||||
"sourceSeq": nullableInt(sourceSeq),
|
||||
"sourceEventId": nullableString(sourceEventID),
|
||||
"terminal": nullableBool(terminal),
|
||||
"sealed": nullableBool(sealed),
|
||||
"createdAt": nullableTime(createdAt),
|
||||
"updatedAt": first(nullableTime(updatedAt), nullableTime(createdAt)),
|
||||
"valuesPrinted": false,
|
||||
"valuesRedacted": true,
|
||||
}
|
||||
if providerProfile.Valid && strings.TrimSpace(providerProfile.String) != "" {
|
||||
item["providerProfile"] = providerProfile.String
|
||||
item["sessionJson"] = map[string]any{"providerProfile": providerProfile.String, "valuesRedacted": true, "secretMaterialStored": false}
|
||||
}
|
||||
result = append(result, item)
|
||||
}
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
func (s *Server) queryAgentTraceEvents(ctx context.Context, query traceEventsQuery) ([]any, error) {
|
||||
clauses := []string{}
|
||||
args := []any{}
|
||||
addTextClause(&clauses, &args, "trace_id", query.TraceID)
|
||||
addTextClause(&clauses, &args, "agent_session_id", query.SessionID)
|
||||
addTextClause(&clauses, &args, "worker_session_id", query.WorkerSessionID)
|
||||
addTextClause(&clauses, &args, "level", query.Level)
|
||||
sqlText := fmt.Sprintf("SELECT event_json FROM agent_trace_events%s ORDER BY occurred_at ASC, id ASC", whereSQL(clauses))
|
||||
if limit := boundedLimit(query.Limit, 0, 1000); limit > 0 {
|
||||
args = append(args, limit)
|
||||
sqlText += fmt.Sprintf(" LIMIT $%d", len(args))
|
||||
}
|
||||
return s.queryJSONColumn(ctx, sqlText, args)
|
||||
}
|
||||
|
||||
func (s *Server) readProjectionOutbox(ctx context.Context, query outboxQuery) ([]any, error) {
|
||||
limit := boundedLimit(query.Limit, 100, 500)
|
||||
args := []any{query.AfterSeq}
|
||||
clauses := []string{"outbox_seq > $1"}
|
||||
addTextClause(&clauses, &args, "trace_id", query.TraceID)
|
||||
addTextClause(&clauses, &args, "session_id", query.SessionID)
|
||||
args = append(args, limit)
|
||||
sqlText := fmt.Sprintf("SELECT outbox_seq, trace_id, session_id, turn_id, message_id, projected_seq, source_seq, source_event_id, commit_type, terminal, sealed, payload_json, created_at FROM workbench_projection_outbox WHERE %s ORDER BY outbox_seq ASC LIMIT $%d", strings.Join(clauses, " AND "), len(args))
|
||||
rows, err := s.db.QueryContext(ctx, sqlText, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
result := []any{}
|
||||
for rows.Next() {
|
||||
var outboxSeq, projectedSeq, sourceSeq sql.NullInt64
|
||||
var traceID, sessionID, turnID, messageID, sourceEventID, commitType sql.NullString
|
||||
var terminal, sealed sql.NullBool
|
||||
var raw []byte
|
||||
var createdAt sql.NullTime
|
||||
if err := rows.Scan(&outboxSeq, &traceID, &sessionID, &turnID, &messageID, &projectedSeq, &sourceSeq, &sourceEventID, &commitType, &terminal, &sealed, &raw, &createdAt); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var payload any
|
||||
if len(raw) > 0 {
|
||||
_ = json.Unmarshal(raw, &payload)
|
||||
}
|
||||
result = append(result, map[string]any{
|
||||
"outboxSeq": nullableInt(outboxSeq),
|
||||
"traceId": nullableString(traceID),
|
||||
"sessionId": nullableString(sessionID),
|
||||
"turnId": nullableString(turnID),
|
||||
"messageId": nullableString(messageID),
|
||||
"projectedSeq": nullableInt(projectedSeq),
|
||||
"sourceSeq": nullableInt(sourceSeq),
|
||||
"sourceEventId": nullableString(sourceEventID),
|
||||
"commitType": nullableString(commitType),
|
||||
"terminal": nullableBool(terminal),
|
||||
"sealed": nullableBool(sealed),
|
||||
"payload": payload,
|
||||
"createdAt": nullableTime(createdAt),
|
||||
"valuesPrinted": false,
|
||||
})
|
||||
}
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
func factClauses(query factQuery, fields []fieldColumn) ([]string, []any) {
|
||||
clauses := []string{}
|
||||
args := []any{}
|
||||
for _, field := range fields {
|
||||
value := singularField(query, field.field)
|
||||
addTextClause(&clauses, &args, field.column, value)
|
||||
addTextListClause(&clauses, &args, field.column, field.values)
|
||||
}
|
||||
return clauses, args
|
||||
}
|
||||
|
||||
func addTextClause(clauses *[]string, args *[]any, column string, value string) {
|
||||
text := strings.TrimSpace(value)
|
||||
if text == "" {
|
||||
return
|
||||
}
|
||||
*args = append(*args, text)
|
||||
*clauses = append(*clauses, fmt.Sprintf("%s = $%d", column, len(*args)))
|
||||
}
|
||||
|
||||
func addTextListClause(clauses *[]string, args *[]any, column string, values []string) {
|
||||
cleaned := []string{}
|
||||
for _, value := range values {
|
||||
if text := strings.TrimSpace(value); text != "" {
|
||||
cleaned = append(cleaned, text)
|
||||
}
|
||||
}
|
||||
if len(cleaned) == 0 {
|
||||
return
|
||||
}
|
||||
placeholders := []string{}
|
||||
for _, value := range cleaned {
|
||||
*args = append(*args, value)
|
||||
placeholders = append(placeholders, fmt.Sprintf("$%d", len(*args)))
|
||||
}
|
||||
*clauses = append(*clauses, fmt.Sprintf("%s IN (%s)", column, strings.Join(placeholders, ", ")))
|
||||
}
|
||||
|
||||
func singularField(query factQuery, field string) string {
|
||||
switch field {
|
||||
case "sessionId": return query.SessionID
|
||||
case "traceId": return query.TraceID
|
||||
case "messageId": return query.MessageID
|
||||
case "turnId": return query.TurnID
|
||||
case "status": return query.Status
|
||||
case "ownerUserId": return query.OwnerUserID
|
||||
case "conversationId": return query.ConversationID
|
||||
case "eventType": return query.EventType
|
||||
case "projectionStatus": return query.ProjectionStatus
|
||||
case "projectionHealth": return query.ProjectionHealth
|
||||
case "runId": return query.RunID
|
||||
case "commandId": return query.CommandID
|
||||
default: return ""
|
||||
}
|
||||
}
|
||||
|
||||
func whereSQL(clauses []string) string {
|
||||
if len(clauses) == 0 {
|
||||
return ""
|
||||
}
|
||||
return " WHERE " + strings.Join(clauses, " AND ")
|
||||
}
|
||||
|
||||
func emptyFacts() map[string][]any {
|
||||
return map[string][]any{
|
||||
"sessions": []any{},
|
||||
"messages": []any{},
|
||||
"parts": []any{},
|
||||
"turns": []any{},
|
||||
"traceEvents": []any{},
|
||||
"checkpoints": []any{},
|
||||
}
|
||||
}
|
||||
|
||||
func factFamilySet(query factQuery) map[string]bool {
|
||||
all := []string{"sessions", "messages", "parts", "turns", "traceEvents", "checkpoints"}
|
||||
selected := query.Families
|
||||
if len(selected) == 0 {
|
||||
selected = query.FactFamilies
|
||||
}
|
||||
result := map[string]bool{}
|
||||
if len(selected) == 0 {
|
||||
for _, family := range all { result[family] = true }
|
||||
return result
|
||||
}
|
||||
valid := map[string]bool{}
|
||||
for _, family := range all { valid[family] = true }
|
||||
for _, family := range selected {
|
||||
family = strings.TrimSpace(family)
|
||||
if valid[family] { result[family] = true }
|
||||
}
|
||||
if len(result) == 0 {
|
||||
for _, family := range all { result[family] = true }
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func factFamilyForTable(table string) string {
|
||||
switch table {
|
||||
case "workbench_sessions": return "sessions"
|
||||
case "workbench_messages": return "messages"
|
||||
case "workbench_parts": return "parts"
|
||||
case "workbench_turns": return "turns"
|
||||
case "workbench_trace_events": return "traceEvents"
|
||||
case "workbench_projection_checkpoints": return "checkpoints"
|
||||
default: return "unknown"
|
||||
}
|
||||
}
|
||||
|
||||
func factOrder(query factQuery, family string) string {
|
||||
value := query.Order
|
||||
switch family {
|
||||
case "sessions": if query.SessionsOrder != "" { value = query.SessionsOrder }
|
||||
case "messages": if query.MessagesOrder != "" { value = query.MessagesOrder }
|
||||
case "parts": if query.PartsOrder != "" { value = query.PartsOrder }
|
||||
case "turns": if query.TurnsOrder != "" { value = query.TurnsOrder }
|
||||
case "checkpoints": if query.CheckpointsOrder != "" { value = query.CheckpointsOrder }
|
||||
}
|
||||
if value == "updated_desc" { return "updated_desc" }
|
||||
return "updated_asc"
|
||||
}
|
||||
|
||||
func sessionSummarySelectClause() string {
|
||||
return strings.Join([]string{
|
||||
"session_id",
|
||||
"owner_user_id",
|
||||
"project_id",
|
||||
"conversation_id",
|
||||
"thread_id",
|
||||
"status",
|
||||
"last_trace_id",
|
||||
"projected_seq",
|
||||
"source_seq",
|
||||
"source_event_id",
|
||||
"terminal",
|
||||
"sealed",
|
||||
"created_at",
|
||||
"updated_at",
|
||||
"COALESCE(NULLIF(session_json::jsonb ->> 'providerProfile', ''), NULLIF(session_json::jsonb #>> '{sessionJson,providerProfile}', '')) AS provider_profile",
|
||||
}, ", ")
|
||||
}
|
||||
|
||||
func boundedLimit(value int, fallback int, max int) int {
|
||||
if value <= 0 {
|
||||
return fallback
|
||||
}
|
||||
if max > 0 && value > max {
|
||||
return max
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
func configFromEnv() Config {
|
||||
port := first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_PORT"), os.Getenv("PORT"), "6671")
|
||||
return Config{
|
||||
Addr: ":" + port,
|
||||
DatabaseURL: first(os.Getenv("HWLAB_WORKBENCH_RUNTIME_DB_URL"), os.Getenv("HWLAB_CLOUD_DB_URL")),
|
||||
QueryTimeout: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_QUERY_TIMEOUT_MS", 10000)) * time.Millisecond,
|
||||
ReadinessDelay: time.Duration(envInt("HWLAB_WORKBENCH_RUNTIME_READY_TIMEOUT_MS", 2000)) * time.Millisecond,
|
||||
PoolMax: envInt("HWLAB_WORKBENCH_RUNTIME_DB_POOL_MAX", 8),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Server) persistenceSummary() map[string]any {
|
||||
return map[string]any{"ready": true, "durable": true, "adapter": "postgres", "serviceId": serviceID, "valuesRedacted": true}
|
||||
}
|
||||
|
||||
func decodeJSON(w http.ResponseWriter, r *http.Request, target any) bool {
|
||||
defer r.Body.Close()
|
||||
decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20))
|
||||
if err := decoder.Decode(target); err != nil {
|
||||
writeAPIError(w, http.StatusBadRequest, "invalid_json", "request body must be valid JSON")
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func writeQueryError(w http.ResponseWriter, err error) {
|
||||
log.Printf("%s query failed: %v", serviceID, err)
|
||||
writeJSON(w, http.StatusServiceUnavailable, map[string]any{
|
||||
"ok": false,
|
||||
"error": map[string]any{
|
||||
"code": "workbench_runtime_query_failed",
|
||||
"message": "workbench runtime query failed",
|
||||
"retryable": true,
|
||||
"transient": true,
|
||||
"causeCode": sqlErrorCode(err),
|
||||
"valuesRedacted": true,
|
||||
},
|
||||
"valuesRedacted": true,
|
||||
})
|
||||
}
|
||||
|
||||
func writeAPIError(w http.ResponseWriter, status int, code string, message string) {
|
||||
writeJSON(w, status, map[string]any{"ok": false, "error": map[string]any{"code": code, "message": message, "valuesRedacted": true}, "valuesRedacted": true})
|
||||
}
|
||||
|
||||
func methodNotAllowed(w http.ResponseWriter, allow string) {
|
||||
w.Header().Set("Allow", allow)
|
||||
writeAPIError(w, http.StatusMethodNotAllowed, "method_not_allowed", "method not allowed")
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, value any) {
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(value)
|
||||
}
|
||||
|
||||
func envInt(name string, fallback int) int {
|
||||
value := strings.TrimSpace(os.Getenv(name))
|
||||
if value == "" { return fallback }
|
||||
parsed, err := strconv.Atoi(value)
|
||||
if err != nil || parsed <= 0 { return fallback }
|
||||
return parsed
|
||||
}
|
||||
|
||||
func first(values ...string) string {
|
||||
for _, value := range values {
|
||||
if strings.TrimSpace(value) != "" { return value }
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func nullableString(value sql.NullString) any {
|
||||
if value.Valid { return value.String }
|
||||
return nil
|
||||
}
|
||||
|
||||
func nullableStringValue(value sql.NullString) string {
|
||||
if value.Valid { return value.String }
|
||||
return ""
|
||||
}
|
||||
|
||||
func nullableInt(value sql.NullInt64) any {
|
||||
if value.Valid { return value.Int64 }
|
||||
return int64(0)
|
||||
}
|
||||
|
||||
func nullableBool(value sql.NullBool) bool {
|
||||
return value.Valid && value.Bool
|
||||
}
|
||||
|
||||
func nullableTime(value sql.NullTime) string {
|
||||
if value.Valid { return value.Time.UTC().Format(time.RFC3339Nano) }
|
||||
return ""
|
||||
}
|
||||
|
||||
type sqlStateError interface { SQLState() string }
|
||||
|
||||
func sqlErrorCode(err error) string {
|
||||
if err == nil { return "" }
|
||||
var stateErr sqlStateError
|
||||
if errors.As(err, &stateErr) { return stateErr.SQLState() }
|
||||
return ""
|
||||
}
|
||||
Reference in New Issue
Block a user