package store import ( "encoding/json" "fmt" "net" "net/http" "strconv" "strings" "time" "github.com/shiv/internal/events" "github.com/shiv/internal/logger" ) // Transaction is a matched HTTP request/response pair. // Host is always a pure hostname — never host:port. // Port holds the destination port as an integer. type Transaction struct { ID uint64 Timestamp time.Time Host string Port int Method string URL string Proto string // "HTTP/2" or "HTTP/1.1" ReqHeaders http.Header ReqBody []byte StatusCode int RespHeaders http.Header RespBody []byte RespSize int DurationMs int64 TLS bool InScope bool } // TransactionFilter holds filter criteria for paginated history queries. // Zero values mean "no filter" for that field. type TransactionFilter struct { Search string // matched against host + url + method + status (case-insensitive) Host string // exact host match (from site map selection) Port int // exact port match; 0 means no filter PathPrefix string // URL path prefix filter (from site map selection) ShowOutScope bool // if false, only in_scope=1 rows are returned Methods []string // whitelist of HTTP methods; empty = all StatusMin int // lower bound on status_code; 0 = no lower bound StatusMax int // upper bound on status_code; 0 = no upper bound HideStatic bool // if true, filter out common static asset extensions in memory } // PageSize is the number of rows returned per page. const PageSize = 100 // SiteMapEntry is a minimal host+port+url triplet used to build the site map // at startup. Port is required to differentiate apps on the same host at // different ports (e.g. localhost:3000 vs localhost:8080). type SiteMapEntry struct { Host string Port int URL string } // SiteMapEntries returns the host, port, and url for every row in history. // Used once at startup to populate the site map with all historical // paths before the first page of transactions is displayed. func (s *Store) SiteMapEntries() ([]SiteMapEntry, error) { rows, err := s.db.Query(`SELECT host, port, url FROM history ORDER BY id ASC`) if err != nil { return nil, fmt.Errorf("store: site map entries: %w", err) } defer rows.Close() var entries []SiteMapEntry for rows.Next() { var e SiteMapEntry if err := rows.Scan(&e.Host, &e.Port, &e.URL); err != nil { return nil, fmt.Errorf("store: scan site map entry: %w", err) } entries = append(entries, e) } return entries, rows.Err() } func (s *Store) GetTransaction(id uint64) (*Transaction, error) { var transaction Transaction var timestampStr, reqHeaders, respHeaders string var tlsFlag, scopeFlag int err := s.db.QueryRow(` SELECT id, timestamp, host, port, method, url, proto, req_headers, req_body, status_code, resp_headers, resp_body, duration_ms, tls, in_scope, resp_size FROM history WHERE id = ?`, id, ).Scan( &transaction.ID, ×tampStr, &transaction.Host, &transaction.Port, &transaction.Method, &transaction.URL, &transaction.Proto, &reqHeaders, &transaction.ReqBody, &transaction.StatusCode, &respHeaders, &transaction.RespBody, &transaction.DurationMs, &tlsFlag, &scopeFlag, &transaction.RespSize, ) if err != nil { return nil, fmt.Errorf("store: get transaction %d: %w", id, err) } transaction.Timestamp, _ = time.Parse(time.RFC3339, timestampStr) transaction.TLS = tlsFlag == 1 transaction.InScope = scopeFlag == 1 if err := json.Unmarshal([]byte(reqHeaders), &transaction.ReqHeaders); err != nil { transaction.ReqHeaders = http.Header{} } if err := json.Unmarshal([]byte(respHeaders), &transaction.RespHeaders); err != nil { transaction.RespHeaders = http.Header{} } return &transaction, nil } // Log writes every transaction as a unique row — no deduplication. func (s *Store) Log(t Transaction) error { return s.write(func() error { reqH, err := json.Marshal(t.ReqHeaders) if err != nil { return fmt.Errorf("store: marshal req headers: %w", err) } respH, err := json.Marshal(t.RespHeaders) if err != nil { return fmt.Errorf("store: marshal resp headers: %w", err) } proto := t.Proto if proto == "" { proto = "HTTP/1.1" } respSize := len(t.RespBody) res, err := s.db.Exec(` INSERT INTO history (timestamp, host, port, method, url, proto, req_headers, req_body, status_code, resp_headers, resp_body, duration_ms, tls, in_scope, resp_size) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, t.Timestamp.UTC().Format(time.RFC3339), t.Host, t.Port, t.Method, t.URL, proto, string(reqH), t.ReqBody, t.StatusCode, string(respH), t.RespBody, t.DurationMs, boolToInt(t.TLS), boolToInt(t.InScope), respSize, ) if err != nil { return fmt.Errorf("store: log transaction: %w", err) } id, _ := res.LastInsertId() t.ID = uint64(id) t.Proto = proto t.RespSize = respSize // Guard against sending on a closed channel when the store is shutting down. select { case <-s.done: default: select { case s.Updates <- t: default: } } return nil }) } // TransactionsPage returns up to PageSize transactions with id < beforeID, // ordered by id descending (newest first), applying filter criteria. // Pass beforeID=0 to start from the newest row. // Bodies are omitted; use GetTransaction for the full row. func (s *Store) TransactionsPage(beforeID uint64, filter TransactionFilter) ([]Transaction, error) { var conditions []string var args []any if beforeID > 0 { conditions = append(conditions, "id < ?") args = append(args, beforeID) } if !filter.ShowOutScope { conditions = append(conditions, "in_scope = 1") } if filter.Host != "" { conditions = append(conditions, "host = ?") args = append(args, filter.Host) if filter.Port != 0 { conditions = append(conditions, "port = ?") args = append(args, filter.Port) } } if filter.PathPrefix != "" { conditions = append(conditions, "(url LIKE ? OR url LIKE ? OR url LIKE ?)") prefix := filter.PathPrefix args = append(args, "%"+prefix, "%"+prefix+"?%", "%"+prefix+"/%", ) } if filter.Search != "" { terms := strings.Fields(strings.ToLower(filter.Search)) for _, term := range terms { conditions = append(conditions, "(LOWER(host) LIKE ? OR LOWER(url) LIKE ? OR LOWER(method) LIKE ? OR CAST(status_code AS TEXT) LIKE ?)") like := "%" + term + "%" args = append(args, like, like, like, like) } } if len(filter.Methods) > 0 { placeholders := make([]string, len(filter.Methods)) for i, m := range filter.Methods { placeholders[i] = "?" args = append(args, strings.ToUpper(m)) } conditions = append(conditions, "method IN ("+strings.Join(placeholders, ",")+")") } if filter.StatusMin > 0 { conditions = append(conditions, "status_code >= ?") args = append(args, filter.StatusMin) } if filter.StatusMax > 0 { conditions = append(conditions, "status_code <= ?") args = append(args, filter.StatusMax) } query := ` SELECT id, timestamp, host, port, method, url, proto, req_headers, '' as req_body, status_code, resp_headers, '' as resp_body, duration_ms, tls, in_scope, resp_size FROM history` if len(conditions) > 0 { query += " WHERE " + strings.Join(conditions, " AND ") } query += fmt.Sprintf(" ORDER BY id DESC LIMIT %d", PageSize) rows, err := s.db.Query(query, args...) if err != nil { return nil, fmt.Errorf("store: query history page: %w", err) } defer rows.Close() return scanTransactions(rows) } // SearchTransactions performs a full-text LIKE scan across request/response bodies // and/or headers. At least one of inReqBody, inRespBody, inHeaders must be true. // Results are ordered by id descending and capped at 200 rows. func (s *Store) SearchTransactions(term string, inReqBody, inRespBody, inHeaders bool) ([]Transaction, error) { if term == "" || (!inReqBody && !inRespBody && !inHeaders) { return nil, nil } like := "%" + term + "%" var clauses []string var args []any if inReqBody { clauses = append(clauses, "CAST(req_body AS TEXT) LIKE ?") args = append(args, like) } if inRespBody { clauses = append(clauses, "CAST(resp_body AS TEXT) LIKE ?") args = append(args, like) } if inHeaders { clauses = append(clauses, "req_headers LIKE ?") args = append(args, like) clauses = append(clauses, "resp_headers LIKE ?") args = append(args, like) } query := ` SELECT id, timestamp, host, port, method, url, proto, req_headers, '' as req_body, status_code, resp_headers, '' as resp_body, duration_ms, tls, in_scope, resp_size FROM history WHERE (` + strings.Join(clauses, " OR ") + `) ORDER BY id DESC LIMIT 200` rows, err := s.db.Query(query, args...) if err != nil { return nil, fmt.Errorf("store: search transactions: %w", err) } defer rows.Close() return scanTransactions(rows) } // AllTransactions returns all transactions ordered by id descending, capped at // 500 rows. Used primarily in tests; production UI uses TransactionsPage. func (s *Store) AllTransactions() ([]Transaction, error) { rows, err := s.db.Query(` SELECT id, timestamp, host, port, method, url, proto, req_headers, req_body, status_code, resp_headers, resp_body, duration_ms, tls, in_scope, resp_size FROM history ORDER BY id DESC LIMIT 500`) if err != nil { return nil, fmt.Errorf("store: query all transactions: %w", err) } defer rows.Close() return scanTransactions(rows) } // ClearHistory deletes all rows from the history table. func (s *Store) ClearHistory() error { return s.write(func() error { _, err := s.db.Exec(`DELETE FROM history`) if err != nil { return fmt.Errorf("store: clear history: %w", err) } _, err = s.db.Exec(`VACUUM`) if err != nil { return fmt.Errorf("store: VACUUM history: %w", err) } return nil }) } func scanTransactions(rows interface { Next() bool Scan(...any) error Err() error }) ([]Transaction, error) { var txs []Transaction for rows.Next() { var tx Transaction var ts, reqH, respH string var tlsInt, scopeInt int if err := rows.Scan( &tx.ID, &ts, &tx.Host, &tx.Port, &tx.Method, &tx.URL, &tx.Proto, &reqH, &tx.ReqBody, &tx.StatusCode, &respH, &tx.RespBody, &tx.DurationMs, &tlsInt, &scopeInt, &tx.RespSize, ); err != nil { return nil, fmt.Errorf("store: scan transaction: %w", err) } tx.Timestamp, _ = time.Parse(time.RFC3339, ts) tx.TLS = tlsInt == 1 tx.InScope = scopeInt == 1 if err := json.Unmarshal([]byte(reqH), &tx.ReqHeaders); err != nil { tx.ReqHeaders = http.Header{} } if err := json.Unmarshal([]byte(respH), &tx.RespHeaders); err != nil { tx.RespHeaders = http.Header{} } txs = append(txs, tx) } return txs, rows.Err() } // ObserveResponse implements events.ResponseObserver. It splits e.Host // (which arrives as host:port from the proxy) into a pure hostname and integer // port once here — no downstream consumer ever parses host/port again. func (s *Store) ObserveResponse(e events.ResponseEvent) { host, portStr, err := net.SplitHostPort(e.Host) if err != nil { host = e.Host if e.TLS { portStr = "443" } else { portStr = "80" } } port, _ := strconv.Atoi(portStr) if err := s.Log(Transaction{ Timestamp: e.Timestamp, Host: host, Port: port, Proto: e.Proto, Method: e.Method, URL: e.URL, ReqHeaders: e.ReqHeaders, ReqBody: e.ReqBody, StatusCode: e.StatusCode, RespHeaders: e.RespHeaders, RespBody: e.RespBody, DurationMs: e.DurationMs, TLS: e.TLS, InScope: s.InScope(host), }); err != nil { logger.Error("store: observe response: %v", err) } } func boolToInt(b bool) int { if b { return 1 } return 0 } // DeleteTransactionsByHostPort deletes all history rows for the given host and port. func (s *Store) DeleteTransactionsByHostPort(host string, port int) error { return s.write(func() error { _, err := s.db.Exec(`DELETE FROM history WHERE host = ? AND port = ?`, host, port) if err != nil { return fmt.Errorf("store: delete transactions for %s:%d: %w", host, port, err) } return nil }) }