From 62ddd06df67edc676b0c878485ae44183d4f02a6 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Mon, 24 Aug 2026 19:08:28 +0800 Subject: [PATCH] Add structured thread lineage endpoint --- README.md | 56 ++++++ cmd/fmsg-webapi/main.go | 1 + internal/handlers/thread.go | 304 ++++++++++++++++++++++++++++++- internal/handlers/thread_test.go | 54 +++++- 4 files changed, 412 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index c2d5a00..c369651 100644 --- a/README.md +++ b/README.md @@ -277,6 +277,8 @@ the application. | `POST` | `/fmsg/:id/read` | Mark a message as read | | `POST` | `/fmsg/:id/add-to` | Add recipients | | `GET` | `/fmsg/:id/data` | Download message data | +| `GET` | `/fmsg/:id/thread` | Render direct ancestry as plain text | +| `GET` | `/fmsg/:id/thread/messages` | Load direct ancestry as structured JSON | | `POST` | `/fmsg/:id/attach` | Upload an attachment | | `GET` | `/fmsg/:id/attach/:filename`| Download an attachment | | `DELETE` | `/fmsg/:id/attach/:filename`| Delete an attachment | @@ -708,6 +710,60 @@ what came before); non-text bodies appear as a `[non-text message: , | `404` | Message not found | | `403` | Authenticated user is not a participant of the requested message | +### GET `/fmsg/:id/thread/messages` + +Returns the requested message and its direct `pid` ancestors as structured +JSON, ordered from the root to the requested message. Sibling branches are not +included. Valid UTF-8 `text/*`, `application/json`, and `application/*+json` +bodies are included inline. Binary bodies and attachments are represented by +authenticated download paths so clients can fetch them concurrently. + +Each visible message includes its normal protocol metadata, a body descriptor, +attachment descriptors, and the canonical message SHA-256 when available. +Because that digest covers the complete message including attachment data, +clients may use the supplied per-part `cache_key` values for content-addressed +caching. Locally delivered messages without a persisted canonical digest are +returned with `cacheable: false`. + +The authenticated identity must be a participant of the requested message. +Ancestors it cannot read appear only as `{"id": ..., "visible": false}` and +make the top-level `complete` field false. The walk is capped at 100 messages +and textual bodies are capped at 32 MiB in aggregate; neither limit is silently +truncated. + +**Response:** `200 OK`, `application/json`: + +```json +{ + "root_id": 41, + "trigger_id": 43, + "complete": true, + "messages": [ + { + "id": 41, + "visible": true, + "pid": null, + "from": "@alice@example.com", + "to": ["@agent@example.net"], + "type": "text/plain", + "size": 5, + "message_sha256": "0123456789abcdef", + "body": {"type": "text/plain", "size": 5, "text": "hello", "cache_key": "sha256:0123456789abcdef:body", "cacheable": true}, + "attachments": [] + } + ] +} +``` + +**Errors:** + +| Status | Condition | +| ------ | --------- | +| `403` | Authenticated user is not a participant of the requested message | +| `404` | Requested message does not exist | +| `413` | Aggregate inline text exceeds 32 MiB (`thread_too_large`) | +| `422` | Direct ancestry exceeds 100 messages (`thread_too_deep`) | + ### POST `/fmsg/:id/attach` Uploads a file attachment for a draft message. Only the owner may upload, and the message must not have been sent. diff --git a/cmd/fmsg-webapi/main.go b/cmd/fmsg-webapi/main.go index 4fa2df3..a614c18 100644 --- a/cmd/fmsg-webapi/main.go +++ b/cmd/fmsg-webapi/main.go @@ -194,6 +194,7 @@ func main() { fmsg.POST("/:id/add-to", msgHandler.AddRecipients) fmsg.GET("/:id/data", msgHandler.DownloadData) fmsg.GET("/:id/thread", msgHandler.ThreadText) + fmsg.GET("/:id/thread/messages", msgHandler.ThreadMessages) fmsg.POST("/:id/attach", attHandler.Upload) fmsg.GET("/:id/attach/:filename", attHandler.Download) diff --git a/internal/handlers/thread.go b/internal/handlers/thread.go index b174f4f..e120458 100644 --- a/internal/handlers/thread.go +++ b/internal/handlers/thread.go @@ -1,22 +1,215 @@ package handlers import ( + "context" "errors" "fmt" "log" "net/http" + "net/url" "os" + "strconv" "strings" "time" + "unicode/utf8" "github.com/gin-gonic/gin" "github.com/jackc/pgx/v5" + + "github.com/markmnl/fmsg-webapi/internal/models" ) // threadMaxHops bounds the ancestor walk (defensive; a pid cycle cannot be // created through the API but the cap keeps the query finite regardless). const threadMaxHops = 100 +const defaultThreadTextBytes int64 = 32 << 20 + +type threadBody struct { + Type string `json:"type"` + Size int `json:"size"` + Text *string `json:"text,omitempty"` + Download string `json:"download,omitempty"` + CacheKey string `json:"cache_key,omitempty"` + Cacheable bool `json:"cacheable"` +} + +type threadAttachment struct { + Position int `json:"position"` + Type string `json:"type"` + Filename string `json:"filename"` + Size int `json:"size"` + Download string `json:"download"` + CacheKey string `json:"cache_key,omitempty"` + Cacheable bool `json:"cacheable"` +} + +type threadMessage struct { + ID int64 `json:"id"` + Visible bool `json:"visible"` + Version int `json:"version,omitempty"` + PID *int64 `json:"pid,omitempty"` + NoReply bool `json:"no_reply,omitempty"` + Important bool `json:"important,omitempty"` + Deflate bool `json:"deflate,omitempty"` + From string `json:"from,omitempty"` + To []string `json:"to,omitempty"` + AddTo []models.AddToBatch `json:"add_to,omitempty"` + Time *float64 `json:"time,omitempty"` + Topic string `json:"topic,omitempty"` + Type string `json:"type,omitempty"` + Size int `json:"size,omitempty"` + MessageSHA256 string `json:"message_sha256,omitempty"` + Body *threadBody `json:"body,omitempty"` + Attachments []threadAttachment `json:"attachments,omitempty"` + dataPath string +} + +type threadMessagesResponse struct { + RootID int64 `json:"root_id"` + TriggerID int64 `json:"trigger_id"` + Complete bool `json:"complete"` + Messages []threadMessage `json:"messages"` +} + +func partCacheKey(messageHash, kind string, position int) string { + if messageHash == "" { + return "" + } + if kind == "body" { + return "sha256:" + messageHash + ":body" + } + return "sha256:" + messageHash + ":attachment:" + strconv.Itoa(position) +} + +func threadDownloadPath(id int64, filename string) string { + if filename == "" { + return fmt.Sprintf("/fmsg/%d/data", id) + } + return fmt.Sprintf("/fmsg/%d/attach/%s", id, url.PathEscape(filename)) +} + +func populateThreadBodies(messages []threadMessage, dataDir string, maxTextBytes int64) error { + var textBytes int64 + for i := range messages { + m := &messages[i] + if !m.Visible { + continue + } + key := partCacheKey(m.MessageSHA256, "body", 0) + m.Body = &threadBody{Type: m.Type, Size: m.Size, Download: threadDownloadPath(m.ID, ""), CacheKey: key, Cacheable: key != ""} + if !isTextType(m.Type) { + continue + } + textBytes += int64(m.Size) + if textBytes > maxTextBytes { + return fmt.Errorf("thread text exceeds %d bytes", maxTextBytes) + } + cleanPath, ok := safeDataPath(m.dataPath, dataDir) + if !ok { + return fmt.Errorf("message %d has an invalid data path", m.ID) + } + raw, err := os.ReadFile(cleanPath) + if err != nil { + return fmt.Errorf("read message %d body: %w", m.ID, err) + } + if utf8.Valid(raw) { + text := string(raw) + m.Body.Text = &text + m.Body.Download = "" + } + } + return nil +} + +func loadThreadRelations(ctx context.Context, tx pgx.Tx, messages []threadMessage) error { + var ids []int64 + byID := make(map[int64]*threadMessage) + for i := range messages { + if messages[i].Visible { + ids = append(ids, messages[i].ID) + byID[messages[i].ID] = &messages[i] + } + } + if len(ids) == 0 { + return nil + } + + rows, err := tx.Query(ctx, `SELECT msg_id, addr FROM msg_to WHERE msg_id = ANY($1) ORDER BY msg_id, id`, ids) + if err != nil { + return err + } + for rows.Next() { + var id int64 + var addr string + if err = rows.Scan(&id, &addr); err != nil { + rows.Close() + return err + } + byID[id].To = append(byID[id].To, addr) + } + rows.Close() + if err = rows.Err(); err != nil { + return err + } + + rows, err = tx.Query(ctx, ` + SELECT b.msg_id, b.id, b.add_to_from, b.time_added, a.addr + FROM msg_add_to_batch b + LEFT JOIN msg_add_to a ON a.batch_id = b.id + WHERE b.msg_id = ANY($1) + ORDER BY b.msg_id, b.id, a.id`, ids) + if err != nil { + return err + } + batchIndexes := make(map[int64]int) + for rows.Next() { + var msgID, batchID int64 + var from string + var added float64 + var addr *string + if err = rows.Scan(&msgID, &batchID, &from, &added, &addr); err != nil { + rows.Close() + return err + } + idx, ok := batchIndexes[batchID] + if !ok { + byID[msgID].AddTo = append(byID[msgID].AddTo, models.AddToBatch{BatchID: batchID, AddToFrom: from, Time: added}) + idx = len(byID[msgID].AddTo) - 1 + batchIndexes[batchID] = idx + } + if addr != nil { + byID[msgID].AddTo[idx].To = append(byID[msgID].AddTo[idx].To, *addr) + } + } + rows.Close() + if err = rows.Err(); err != nil { + return err + } + + rows, err = tx.Query(ctx, ` + SELECT msg_id, position, type, filename, filesize + FROM msg_attachment WHERE msg_id = ANY($1) + ORDER BY msg_id, position, filename`, ids) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + var msgID int64 + var a threadAttachment + if err = rows.Scan(&msgID, &a.Position, &a.Type, &a.Filename, &a.Size); err != nil { + return err + } + m := byID[msgID] + a.Download = threadDownloadPath(msgID, a.Filename) + a.CacheKey = partCacheKey(m.MessageSHA256, "attachment", a.Position) + a.Cacheable = a.CacheKey != "" + m.Attachments = append(m.Attachments, a) + } + return rows.Err() +} + // threadEntry is one message on the direct pid lineage, root first. type threadEntry struct { ID int64 @@ -59,8 +252,8 @@ func renderThreadText(entries []threadEntry) string { // isTextType reports whether a MIME type's body can be inlined as text. func isTextType(mimeType string) bool { - t := strings.ToLower(strings.TrimSpace(mimeType)) - return strings.HasPrefix(t, "text/") || t == "application/json" + t := strings.ToLower(strings.TrimSpace(strings.SplitN(mimeType, ";", 2)[0])) + return strings.HasPrefix(t, "text/") || t == "application/json" || strings.HasSuffix(t, "+json") } // ThreadText handles GET /fmsg/:id/thread — returns the message's direct @@ -159,3 +352,110 @@ func (h *MessageHandler) ThreadText(c *gin.Context) { c.Data(http.StatusOK, "text/plain; charset=utf-8", []byte(renderThreadText(entries))) } + +// ThreadMessages handles GET /fmsg/:id/thread/messages. It returns the direct +// pid lineage as structured JSON, root first, with textual bodies inlined and +// binary bodies plus attachments represented by authenticated download paths. +func (h *MessageHandler) ThreadMessages(c *gin.Context) { + addrs, err := h.visibleAddrs(c) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + msgID, ok := parseID(c) + if !ok { + return + } + ctx := c.Request.Context() + tx, err := h.DB.Pool.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly}) + if err != nil { + log.Printf("thread messages: begin: %v", err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + defer tx.Rollback(ctx) + + rows, err := tx.Query(ctx, ` + WITH RECURSIVE chain AS ( + SELECT id, version, pid, no_reply, is_important, is_deflate, time_sent, + from_addr, topic, type, size, filepath, sha256, 0 AS depth + FROM msg WHERE id = $1 + UNION ALL + SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate, + m.time_sent, m.from_addr, m.topic, m.type, m.size, m.filepath, + m.sha256, c.depth + 1 + FROM msg m JOIN chain c ON m.id = c.pid + WHERE c.depth + 1 < $3 + ) + SELECT c.id, c.version, c.pid, c.no_reply, c.is_important, c.is_deflate, + c.time_sent, c.from_addr, c.topic, c.type, c.size, c.filepath, + encode(c.sha256, 'hex'), + (c.from_addr = ANY($2) + OR EXISTS (SELECT 1 FROM msg_to t WHERE t.msg_id = c.id AND t.addr = ANY($2)) + OR EXISTS (SELECT 1 FROM msg_add_to a WHERE a.msg_id = c.id AND a.addr = ANY($2))) + FROM chain c ORDER BY c.depth DESC`, msgID, addrs, threadMaxHops) + if err != nil { + log.Printf("thread messages: walk %d: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + var messages []threadMessage + for rows.Next() { + var m threadMessage + var hash *string + if err = rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, + &m.Deflate, &m.Time, &m.From, &m.Topic, &m.Type, &m.Size, &m.dataPath, + &hash, &m.Visible); err != nil { + rows.Close() + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + if hash != nil { + m.MessageSHA256 = *hash + } + messages = append(messages, m) + } + rows.Close() + if err = rows.Err(); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + if len(messages) == 0 { + c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) + return + } + if !messages[len(messages)-1].Visible { + c.JSON(http.StatusForbidden, gin.H{"error": "access denied"}) + return + } + if len(messages) == threadMaxHops && messages[0].PID != nil { + c.JSON(http.StatusUnprocessableEntity, gin.H{"error": "thread exceeds maximum ancestry", "code": "thread_too_deep"}) + return + } + if err = loadThreadRelations(ctx, tx, messages); err != nil { + log.Printf("thread messages: relations %d: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + if err = tx.Commit(ctx); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + if err = populateThreadBodies(messages, h.DataDir, defaultThreadTextBytes); err != nil { + if strings.Contains(err.Error(), "exceeds") { + c.JSON(http.StatusRequestEntityTooLarge, gin.H{"error": err.Error(), "code": "thread_too_large"}) + return + } + log.Printf("thread messages: bodies %d: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "thread data is unavailable", "code": "thread_data_unavailable"}) + return + } + complete := true + for i := range messages { + if !messages[i].Visible { + complete = false + messages[i] = threadMessage{ID: messages[i].ID, Visible: false} + } + } + c.JSON(http.StatusOK, threadMessagesResponse{RootID: messages[0].ID, TriggerID: msgID, Complete: complete, Messages: messages}) +} diff --git a/internal/handlers/thread_test.go b/internal/handlers/thread_test.go index 4f93fcc..3189aaa 100644 --- a/internal/handlers/thread_test.go +++ b/internal/handlers/thread_test.go @@ -1,6 +1,9 @@ package handlers import ( + "encoding/json" + "os" + "path/filepath" "strings" "testing" ) @@ -33,10 +36,59 @@ func TestRenderThreadText(t *testing.T) { func TestIsTextType(t *testing.T) { for typ, want := range map[string]bool{ "text/plain": true, "text/markdown": true, "TEXT/HTML": true, - "application/json": true, "image/png": false, "application/pdf": false, "": false, + "application/json": true, "application/problem+json": true, + "text/plain; charset=utf-8": true, "image/png": false, "application/pdf": false, "": false, } { if isTextType(typ) != want { t.Errorf("isTextType(%q) != %v", typ, want) } } } + +func TestPopulateThreadBodiesAndCacheKeys(t *testing.T) { + dir := t.TempDir() + textPath := filepath.Join(dir, "text") + if err := os.WriteFile(textPath, []byte("hello"), 0600); err != nil { + t.Fatal(err) + } + messages := []threadMessage{ + {ID: 1, Visible: true, Type: "text/plain", Size: 5, MessageSHA256: "abc", dataPath: textPath}, + {ID: 2, Visible: true, Type: "application/pdf", Size: 9, dataPath: filepath.Join(dir, "pdf")}, + {ID: 3, Visible: false, Type: "text/plain", Size: 999, dataPath: "/outside"}, + } + if err := populateThreadBodies(messages, dir, 5); err != nil { + t.Fatal(err) + } + if messages[0].Body == nil || messages[0].Body.Text == nil || *messages[0].Body.Text != "hello" { + t.Fatalf("text body not inlined: %#v", messages[0].Body) + } + if messages[0].Body.CacheKey != "sha256:abc:body" || !messages[0].Body.Cacheable { + t.Fatalf("unexpected body cache metadata: %#v", messages[0].Body) + } + if messages[1].Body.Download != "/fmsg/2/data" || messages[1].Body.Cacheable { + t.Fatalf("unexpected binary descriptor: %#v", messages[1].Body) + } + if messages[2].Body != nil { + t.Fatal("invisible body must not be populated") + } +} + +func TestThreadMessagesJSONDoesNotExposeInternalPath(t *testing.T) { + r := threadMessagesResponse{RootID: 1, TriggerID: 1, Complete: true, Messages: []threadMessage{{ID: 1, Visible: true, dataPath: "/secret/path"}}} + b, err := json.Marshal(r) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(b), "secret") { + t.Fatalf("internal path leaked: %s", b) + } +} + +func TestThreadAttachmentCacheKey(t *testing.T) { + if got := partCacheKey("deadbeef", "attachment", 4); got != "sha256:deadbeef:attachment:4" { + t.Fatalf("got %q", got) + } + if got := partCacheKey("", "attachment", 4); got != "" { + t.Fatalf("hashless message must not be cacheable: %q", got) + } +}