From 6469d06ea844102bd8988adab6c9a479ae6d5e0c Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Wed, 21 Oct 2026 22:24:59 +0800 Subject: [PATCH 1/2] Trash: delete, restore and delete for good Each identity gets a trash. DELETE /fmsg/:id on a sent message moves it (or with ?thread=true its whole thread, later replies included) into the caller's trash; POST /fmsg/:id/restore takes it out; ?permanent=true deletes it for good. Lists leave trashed messages out unless ?trash=include or only, and every message carries its trash state. A message deleted for good, and everything replying to it, is 404 on every route for that identity, never listed, pushed or sent over the WebSocket, and its usage is released to fmsgid. Once no local participant holds it and no delivery is pending, it is removed from the host, so fmsgd rejects replies to it. Co-Authored-By: Claude Opus 5.5 --- README.md | 68 ++- .../apikey_cli_integration_test.go | 12 + cmd/fmsg-webapi/main.go | 1 + dd.sql | 34 ++ internal/handlers/attachments.go | 2 +- .../handlers/finalization_integration_test.go | 9 + internal/handlers/hub.go | 55 +- internal/handlers/messages.go | 82 ++- internal/handlers/push.go | 3 + internal/handlers/reactions.go | 2 +- internal/handlers/read_event_test.go | 37 ++ internal/handlers/references.go | 5 + .../handlers/subaccounts_lifecycle_test.go | 12 + internal/handlers/thread.go | 42 +- internal/handlers/trash.go | 520 ++++++++++++++++++ internal/handlers/trash_integration_test.go | 311 +++++++++++ internal/middleware/fmsgid_usage.go | 14 + internal/middleware/oauth.go | 1 + internal/models/models.go | 1 + 19 files changed, 1180 insertions(+), 31 deletions(-) create mode 100644 internal/handlers/trash.go create mode 100644 internal/handlers/trash_integration_test.go diff --git a/README.md b/README.md index 8a96314..afbf249 100644 --- a/README.md +++ b/README.md @@ -60,8 +60,9 @@ Standard PostgreSQL environment variables (`PGHOST`, `PGPORT`, `PGUSER`, fmsg-webapi assumes [fmsgd](https://github.com/markmnl/fmsgd) is the fmsg host implementation and uses the same PostgreSQL database. The fmsgd schema remains -the base schema; this repository's [dd.sql](dd.sql) adds the API-key and -delegation tables used by fmsg-webapi. +the base schema; this repository's [dd.sql](dd.sql) adds the API-key, +delegation and [trash](#trash) tables used by fmsg-webapi. Apply it after the +fmsgd schema, which it extends. A `.env` file placed in the working directory is loaded automatically at startup (values in the environment take precedence). @@ -347,9 +348,10 @@ the application. | `POST` | `/fmsg` | Create a draft message | | `GET` | `/fmsg/:id` | Retrieve a message | | `PUT` | `/fmsg/:id` | Update a draft message | -| `DELETE` | `/fmsg/:id` | Delete a draft message | +| `DELETE` | `/fmsg/:id` | Delete a draft; move a sent message to the trash, or delete it for good | | `POST` | `/fmsg/:id/send` | Send a message | | `POST` | `/fmsg/:id/read` | Mark a message as read | +| `POST` | `/fmsg/:id/restore` | Take a message or thread out of the trash | | `POST` | `/fmsg/:id/add-to` | Add recipients | | `POST` | `/fmsg/:id/react` | Set or clear an emoji reaction (FMSG-005) | | `GET` | `/fmsg/:id/data` | Download message data | @@ -554,6 +556,10 @@ event types can be added without breaking clients: | `recipients_added` | The refreshed message object. | An add-to batch was recorded on a message the user participates in. | | `read` | The message object, with `read` and `time_read` set. | The user marked a message read (`POST /fmsg/:id/read`), on this or any other connection, so every tab and device can stop showing it as unread. Sent once, when the message first becomes read. | | `reaction` | The refreshed **subject** message object, with its `reactions` up to date. | A reaction (FMSG-005) on a message the user participates in arrives. The reaction message itself is not pushed as `new_msg` and triggers no Web Push. | +| `trash` | The message object with its `trash` state, or `{"id": , "trash": "purged"}` for a message deleted for good. | The user moved a message into or out of their [trash](#trash), or deleted it for good, on this or any other connection. | + +No event about a message the user deleted for good is ever pushed again, +except that `trash` event. A client only ever receives events for messages it is a participant on. The server sends periodic WebSocket pings; clients should respond with pongs (most @@ -589,6 +595,7 @@ has read the message. | --------- | ------- | ----------- | | `limit` | `20` | Max messages to return (at least 1; values above 100 return 100) | | `offset` | `0` | Number of messages to skip for pagination | +| `trash` | `exclude` | Messages in the caller's [trash](#trash): `exclude` leaves them out, `include` lists them too, `only` lists only them. Messages deleted for good are never listed. | **Response:** JSON array of message objects. Each object has the same shape as the single-message response from `GET /fmsg/:id` (with an additional `id` field), including the `read`/`time_read` fields reflecting the caller's per-recipient read state. Message body data and attachment contents are not included — use the dedicated download endpoints instead. @@ -604,6 +611,7 @@ This includes both sent messages and drafts (`time_sent` may be `NULL`). | --------- | ------- | ----------- | | `limit` | `20` | Max messages to return (at least 1; values above 100 return 100) | | `offset` | `0` | Number of messages to skip for pagination | +| `trash` | `exclude` | As for `GET /fmsg`. | **Response:** JSON array of message objects with the same shape as `GET /fmsg`. @@ -760,16 +768,59 @@ Updates a draft message. Only the owner (`from`) may update, and the message mus ### DELETE `/fmsg/:id` -Deletes a draft message and all its attachments from the database and disk. Only the owner may delete, and the message must not have been sent. +**A draft:** deletes it and all its attachments from the database and disk. Only +the owner may delete a draft. **Response:** `204 No Content`. -**Response:** `204 No Content`. +**A sent message:** moves it into the caller's [trash](#trash). Its sender and +any of its recipients may do this; it changes nothing for anyone else. + +| Parameter | Description | +| --------- | ----------- | +| `thread=true` | Delete the whole thread below this message, including replies that arrive later. | +| `permanent=true` | Delete for good: this message and everything below it (with `thread=true`, the whole thread). | + +**Response:** `200 OK` with `{"id": , "trash": "message" | "thread"}`, or +for `permanent=true` `{"id": , "trash": "purged", "purged": , "removed": }`: +how many messages were newly deleted for good, and how many of those were +removed from the host. **Errors:** | Status | Condition | | ------ | --------- | -| `403` | Not the owner, or message already sent | -| `404` | Message not found | +| `403` | A draft and not its owner | +| `404` | Message not found, the caller is not a participant, or it is already deleted for good | + +### POST `/fmsg/:id/restore` + +Takes a message out of the caller's [trash](#trash). With `thread=true`, takes +the whole thread below it out. Messages deleted for good can't be restored. + +**Response:** `200 OK` with `{"id": , "trash": null}`; `trash` is still +`"thread"` when the message is in a deleted thread that wasn't restored. + +**Errors:** `404` when the message isn't found, the caller is not a +participant, or it is deleted for good. + +### Trash + +Each identity has its own trash; nothing in it changes what anyone else sees. + +- **Deleted** messages (`"trash": "message"`) and messages in a deleted thread + (`"trash": "thread"`) are left out of `GET /fmsg` and `GET /fmsg/sent` unless + `trash=include` or `trash=only` is given, and carry their `trash` state on + every route. Replies keep arriving: a reply in a deleted thread is in the + trash too, and triggers no [Web Push](#web-push). +- **Deleted for good** (`permanent=true`) is final. The message and everything + that replies to it are never listed again for that identity, every route + answers `404` for them (or shows them as not visible in a thread), and a + reply to them is refused (`400 parent message not found`). What they counted + against that identity's storage quota is given back to fmsgid. Once no local + participant holds a message any more (none sent or accepted it without + deleting it for good) and none of its deliveries is pending, it is removed + from the host: its rows and files are deleted, and replies to it from + anywhere are then rejected by fmsgd as parent not found. This needs the + fmsgd schema that lets a transaction set `fmsg.purge`. ### Message hashes and references @@ -1130,7 +1181,8 @@ The push carries a JSON payload: ``` `threadId`, `url` and `tag` reference the thread's **root** message id, so -replies group with their thread. If a push service responds `404` or `410` the +replies group with their thread. A message in the recipient's +[trash](#trash), such as a reply in a thread they deleted, is not pushed. If a push service responds `404` or `410` the subscription is dead and is deleted automatically. **Enabling.** Web Push is active only when `FMSG_VAPID_PUBLIC_KEY`, diff --git a/cmd/fmsg-webapi/apikey_cli_integration_test.go b/cmd/fmsg-webapi/apikey_cli_integration_test.go index d05e65b..3676bb1 100644 --- a/cmd/fmsg-webapi/apikey_cli_integration_test.go +++ b/cmd/fmsg-webapi/apikey_cli_integration_test.go @@ -51,6 +51,18 @@ func newAPIKeyCLIDB(t *testing.T) *db.DB { t.Fatal(err) } t.Cleanup(pool.Close) + // This service's dd.sql extends the fmsgd schema. + fmsgdDD := os.Getenv("FMSG_TEST_DD") + if fmsgdDD == "" { + fmsgdDD = "../../../fmsgd/dd.sql" + } + base, err := os.ReadFile(fmsgdDD) + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, string(base)); err != nil { + t.Fatal(err) + } dd, err := os.ReadFile("../../dd.sql") if err != nil { t.Fatal(err) diff --git a/cmd/fmsg-webapi/main.go b/cmd/fmsg-webapi/main.go index 358973f..3704e26 100644 --- a/cmd/fmsg-webapi/main.go +++ b/cmd/fmsg-webapi/main.go @@ -284,6 +284,7 @@ func registerRoutes(router *gin.Engine, jwtMiddleware gin.HandlerFunc, msgHandle fmsg.DELETE("/:id", msgHandler.Atomic((*handlers.MessageHandler).Delete)) fmsg.POST("/:id/send", msgHandler.Atomic((*handlers.MessageHandler).Send)) fmsg.POST("/:id/read", msgHandler.MarkRead) + fmsg.POST("/:id/restore", msgHandler.Atomic((*handlers.MessageHandler).Restore)) fmsg.POST("/:id/add-to", msgHandler.Atomic((*handlers.MessageHandler).AddRecipients)) fmsg.POST("/:id/react", msgHandler.Atomic((*handlers.MessageHandler).React)) fmsg.GET("/:id/data", msgHandler.DownloadData) diff --git a/dd.sql b/dd.sql index 4f6db9c..73fd2b5 100644 --- a/dd.sql +++ b/dd.sql @@ -72,3 +72,37 @@ CREATE INDEX IF NOT EXISTS fmsg_api_sub_account_sub_idx CREATE UNIQUE INDEX IF NOT EXISTS fmsg_api_sub_account_owner_sub_unique ON fmsg_api_sub_account ((lower(owner_addr)), (lower(sub_addr))) WHERE agent <> ''; + +-- Trash: what each address has deleted. A row marks one message for one +-- address (lower-cased). thread marks the whole thread below that message; +-- time_purged means deleted for good: never shown to that address again, +-- nor anything that replies to it. Rows go with their message when it is +-- removed from the host. +CREATE TABLE IF NOT EXISTS fmsg_api_trash ( + addr text NOT NULL, + msg_id bigint NOT NULL REFERENCES msg (id) ON DELETE CASCADE, + thread boolean NOT NULL DEFAULT false, + time_deleted double precision NOT NULL, + time_purged double precision, + PRIMARY KEY (addr, msg_id) +); + +-- fmsg_api_trash_state is where a message stands for an address: 'purged' +-- when it or an ancestor is deleted for good, 'thread' when it or an ancestor +-- carries a thread mark, 'message' when it is deleted itself, NULL when it +-- isn't deleted. addr must be lower-cased. +CREATE OR REPLACE FUNCTION fmsg_api_trash_state(p_addr text, p_msg bigint) RETURNS text +LANGUAGE sql STABLE AS $$ + WITH RECURSIVE up(id, depth) AS ( + SELECT p_msg, 0 + UNION ALL + SELECT m.pid, up.depth + 1 FROM msg m JOIN up ON m.id = up.id + WHERE m.pid IS NOT NULL AND up.depth < 1000 + ) + SELECT CASE + WHEN bool_or(t.time_purged IS NOT NULL) THEN 'purged' + WHEN bool_or(t.thread) THEN 'thread' + WHEN bool_or(up.depth = 0) THEN 'message' + END + FROM up JOIN fmsg_api_trash t ON t.addr = p_addr AND t.msg_id = up.id +$$; diff --git a/internal/handlers/attachments.go b/internal/handlers/attachments.go index 77ef36d..25adac9 100644 --- a/internal/handlers/attachments.go +++ b/internal/handlers/attachments.go @@ -198,7 +198,7 @@ func (h *AttachmentHandler) Download(c *gin.Context) { } return } - if draftHidden([]string{identity}, fromAddr, timeSent) { + if draftHidden([]string{identity}, fromAddr, timeSent) || purgedFor(ctx, h.q(), identity, msgID) { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } diff --git a/internal/handlers/finalization_integration_test.go b/internal/handlers/finalization_integration_test.go index 64bd01b..a12b9f0 100644 --- a/internal/handlers/finalization_integration_test.go +++ b/internal/handlers/finalization_integration_test.go @@ -75,6 +75,14 @@ func newFinalizationAPIWithID(t *testing.T, fmsgID http.Handler) *finalizationAP if _, err = pool.Exec(ctx, string(sql)); err != nil { t.Fatal(err) } + // This service's own tables (dd.sql at the repository root). + apiSQL, err := os.ReadFile("../../dd.sql") + if err != nil { + t.Fatal(err) + } + if _, err = pool.Exec(ctx, string(apiSQL)); err != nil { + t.Fatal(err) + } id := httptest.NewServer(fmsgID) t.Cleanup(id.Close) h := NewMessageHandler(&db.DB{Pool: pool}, t.TempDir(), 1<<20, 2<<20, 256, nil, id.URL, "example.com") @@ -91,6 +99,7 @@ func newFinalizationAPIWithID(t *testing.T, fmsgID http.Handler) *finalizationAP r.GET("/fmsg/:id/data", h.DownloadData) r.POST("/fmsg/:id/send", h.Atomic((*MessageHandler).Send)) r.POST("/fmsg/:id/read", h.MarkRead) + r.POST("/fmsg/:id/restore", h.Atomic((*MessageHandler).Restore)) r.POST("/fmsg/:id/add-to", h.Atomic((*MessageHandler).AddRecipients)) r.POST("/fmsg/:id/react", h.Atomic((*MessageHandler).React)) r.GET("/fmsg/:id/thread", h.ThreadText) diff --git a/internal/handlers/hub.go b/internal/handlers/hub.go index 6d4f52e..6caf00d 100644 --- a/internal/handlers/hub.go +++ b/internal/handlers/hub.go @@ -20,6 +20,7 @@ const ( eventRecipientsAdded = "recipients_added" eventReaction = "reaction" eventRead = "read" + eventTrash = "trash" ) // wsEnvelope is the JSON shape of every frame pushed over a WebSocket. The @@ -144,7 +145,10 @@ func (h *Hub) listen(ctx context.Context, onConnected func()) error { if _, err := conn.Exec(ctx, "LISTEN msg_read"); err != nil { return err } - log.Println("ws hub: listening on new_msg, delivered, recipients_added, msg_read") + if _, err := conn.Exec(ctx, "LISTEN msg_trash"); err != nil { + return err + } + log.Println("ws hub: listening on new_msg, delivered, recipients_added, msg_read, msg_trash") onConnected() for { @@ -191,6 +195,11 @@ func (h *Hub) route(ctx context.Context, channel string, msgID int64, addr strin // recipient. Pushes the message with its read state so the reader's // other tabs and devices stop showing it as unread. h.dispatch(ctx, msgID, addr, eventRead) + case "msg_trash": + // The address moved the message into or out of its trash, or + // deleted it for good (see trash.go). Its other tabs and devices + // get the message with its trash state. + h.dispatchTrash(ctx, msgID, addr) case "recipients_added": // An add-to batch was recorded against the message; addr is one // of its participants (see notify_recipients_added in fmsgd's @@ -244,7 +253,49 @@ func (h *Hub) dispatch(ctx context.Context, msgID int64, addr string, eventType log.Printf("ws hub: build message %d for %s: %v", msgID, addr, err) return } - payload, err := json.Marshal(wsEnvelope{Type: eventType, Data: item}) + if item.purged { + return // deleted for good by addr: never shown to it again + } + h.send(clients, msgID, wsEnvelope{Type: eventType, Data: item}) +} + +// trashGone is the trash event data for a message deleted for good. +type trashGone struct { + ID int64 `json:"id"` + Trash string `json:"trash"` +} + +// dispatchTrash pushes a trash event to addr's clients: the message with its +// trash state, or only its id when addr deleted it for good (it may no longer +// exist). +func (h *Hub) dispatchTrash(ctx context.Context, msgID int64, addr string) { + clients := h.clientsOf(addr) + if len(clients) == 0 { + return + } + item, err := h.buildItem(ctx, msgID, addr) + if err != nil || item.purged { + h.send(clients, msgID, wsEnvelope{Type: eventTrash, Data: trashGone{ID: msgID, Trash: trashPurged}}) + return + } + h.send(clients, msgID, wsEnvelope{Type: eventTrash, Data: item}) +} + +// clientsOf lists the clients connected as addr. +func (h *Hub) clientsOf(addr string) []*wsClient { + h.mu.RLock() + defer h.mu.RUnlock() + set := h.registry[strings.ToLower(addr)] + clients := make([]*wsClient, 0, len(set)) + for c := range set { + clients = append(clients, c) + } + return clients +} + +// send marshals env and queues it on each client. +func (h *Hub) send(clients []*wsClient, msgID int64, env wsEnvelope) { + payload, err := json.Marshal(env) if err != nil { log.Printf("ws hub: marshal message %d: %v", msgID, err) return diff --git a/internal/handlers/messages.go b/internal/handlers/messages.go index db44f6e..397fd0e 100644 --- a/internal/handlers/messages.go +++ b/internal/handlers/messages.go @@ -90,8 +90,12 @@ type messageListItem struct { Attachments []models.Attachment `json:"attachments"` Reaction *string `json:"reaction"` Reactions []models.Reaction `json:"reactions"` + // Trash is "message" or "thread" when the message is in the caller's + // trash, null otherwise. + Trash *string `json:"trash"` dataPath string // stored body path, for reaction recognition; never serialised + purged bool // deleted for good by the address it was built for; never serialised } // formatDeliveredISO converts a nullable Unix-epoch-seconds timestamp (as @@ -312,22 +316,34 @@ func (h *MessageHandler) List(c *gin.Context) { if !ok { return } + trashFilter, ok := trashListFilter(c) + if !ok { + return + } ctx := c.Request.Context() + hasTrash, err := h.hasTrash(ctx, addrs[0]) + if err != nil { + log.Printf("list messages: trash: %v", err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list messages"}) + return + } rows, err := h.q().Query(ctx, `SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate, m.is_terminal, m.time_sent, encode(m.sha256,'hex'), encode(m.psha256,'hex'), m.from_addr, m.topic, m.type, m.size, m.filepath, COALESCE( (SELECT mt.time_read FROM msg_to mt WHERE mt.msg_id = m.id AND lower(mt.addr) = ANY($1)), (SELECT mat.time_read FROM msg_add_to mat WHERE mat.msg_id = m.id AND lower(mat.addr) = ANY($1)) - ) AS time_read + ) AS time_read, t.state FROM msg m + CROSS JOIN LATERAL (SELECT CASE WHEN $4 THEN fmsg_api_trash_state($5, m.id) END AS state) t WHERE m.time_sent IS NOT NULL AND (EXISTS (SELECT 1 FROM msg_to mt WHERE mt.msg_id = m.id AND lower(mt.addr) = ANY($1)) OR EXISTS (SELECT 1 FROM msg_add_to mat WHERE mat.msg_id = m.id AND lower(mat.addr) = ANY($1))) + AND `+trashFilter+` ORDER BY m.id DESC LIMIT $2 OFFSET $3`, - addrs, limit, offset, + addrs, limit, offset, hasTrash, addrs[0], ) if err != nil { log.Printf("list messages: %v", err) @@ -341,7 +357,7 @@ func (h *MessageHandler) List(c *gin.Context) { for rows.Next() { var m messageListItem var dataPath string - if err := rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.SHA256, &m.PSHA256, &m.From, &m.Topic, &m.Type, &m.Size, &dataPath, &m.TimeRead); err != nil { + if err := rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.SHA256, &m.PSHA256, &m.From, &m.Topic, &m.Type, &m.Size, &dataPath, &m.TimeRead, &m.Trash); err != nil { log.Printf("list messages scan: %v", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list messages"}) return @@ -417,16 +433,28 @@ func (h *MessageHandler) Sent(c *gin.Context) { if !ok { return } + trashFilter, ok := trashListFilter(c) + if !ok { + return + } ctx := c.Request.Context() + hasTrash, err := h.hasTrash(ctx, addrs[0]) + if err != nil { + log.Printf("list sent messages: trash: %v", err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list sent messages"}) + return + } rows, err := h.q().Query(ctx, - `SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate, m.is_terminal, m.time_sent, encode(m.sha256,'hex'), encode(m.psha256,'hex'), m.from_addr, m.topic, m.type, m.size, m.filepath + `SELECT m.id, m.version, m.pid, m.no_reply, m.is_important, m.is_deflate, m.is_terminal, m.time_sent, encode(m.sha256,'hex'), encode(m.psha256,'hex'), m.from_addr, m.topic, m.type, m.size, m.filepath, t.state FROM msg m + CROSS JOIN LATERAL (SELECT CASE WHEN $4 AND m.time_sent IS NOT NULL THEN fmsg_api_trash_state($5, m.id) END AS state) t WHERE lower(m.from_addr) = ANY($1) + AND `+trashFilter+` ORDER BY m.id DESC LIMIT $2 OFFSET $3`, - addrs, limit, offset, + addrs, limit, offset, hasTrash, addrs[0], ) if err != nil { log.Printf("list sent messages: %v", err) @@ -440,7 +468,7 @@ func (h *MessageHandler) Sent(c *gin.Context) { for rows.Next() { var m messageListItem var dataPath string - if err := rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.SHA256, &m.PSHA256, &m.From, &m.Topic, &m.Type, &m.Size, &dataPath); err != nil { + if err := rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.SHA256, &m.PSHA256, &m.From, &m.Topic, &m.Type, &m.Size, &dataPath, &m.Trash); err != nil { log.Printf("list sent messages scan: %v", err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list sent messages"}) return @@ -632,6 +660,19 @@ func (h *MessageHandler) Get(c *gin.Context) { c.JSON(http.StatusForbidden, gin.H{"error": "access denied"}) return } + trash, err := trashStateQ(ctx, h.q(), addrs[0], msgID) + if err != nil { + log.Printf("get message %d: trash: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve message"}) + return + } + if trash == trashPurged { + c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) + return + } + if trash != "" { + msg.Trash = &trash + } // Compute ShortText only after authorization has been confirmed. msg.ShortText = h.extractShortText(dataPath, msg.Type) @@ -692,7 +733,7 @@ func (h *MessageHandler) DownloadData(c *gin.Context) { } return } - if draftHidden(addrs, fromAddr, timeSent) { + if draftHidden(addrs, fromAddr, timeSent) || purgedFor(ctx, h.q(), addrs[0], msgID) { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } @@ -850,7 +891,8 @@ func (h *MessageHandler) Update(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"id": msgID}) } -// Delete handles DELETE /fmsg/:id — deletes a draft message. +// Delete handles DELETE /fmsg/:id — deletes a draft, or moves a sent message +// to the caller's trash (see trash.go). func (h *MessageHandler) Delete(c *gin.Context) { identity := middleware.GetIdentity(c) msgID, ok := resolveID(c, h.q()) @@ -870,12 +912,14 @@ func (h *MessageHandler) Delete(c *gin.Context) { return } - if !sameAddr(existing.From, identity) { - c.JSON(http.StatusForbidden, gin.H{"error": "only the owner may delete a message"}) + // A sent message goes to the caller's trash (or is deleted for good with + // ?permanent=true); only a draft is deleted outright, by its owner. + if existing.Time != nil { + h.trashSent(c, msgID) return } - if existing.Time != nil { - c.JSON(http.StatusForbidden, gin.H{"error": "sent messages cannot be deleted"}) + if !sameAddr(existing.From, identity) { + c.JSON(http.StatusForbidden, gin.H{"error": "only the owner may delete a message"}) return } @@ -1063,7 +1107,7 @@ func (h *MessageHandler) MarkRead(c *gin.Context) { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to mark message read"}) return } - if !recipient { + if !recipient || purgedFor(ctx, h.q(), identity, msgID) { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } @@ -1157,7 +1201,7 @@ func (h *MessageHandler) AddRecipients(c *gin.Context) { } return } - if draftHidden([]string{identity}, fromAddr, timeSent) { + if draftHidden([]string{identity}, fromAddr, timeSent) || purgedFor(ctx, h.q(), identity, msgID) { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } @@ -1504,6 +1548,16 @@ func (h *MessageHandler) messageItemFor(ctx context.Context, msgID int64, recipi item.Read = timeRead != nil } + // Where it stands in the recipient's trash. + trash, err := trashStateQ(ctx, h.q(), recipient, msgID) + if err != nil { + return nil, err + } + item.purged = trash == trashPurged + if trash != "" && !item.purged { + item.Trash = &trash + } + return item, nil } diff --git a/internal/handlers/push.go b/internal/handlers/push.go index 2d2bb8e..3ea03e7 100644 --- a/internal/handlers/push.go +++ b/internal/handlers/push.go @@ -181,6 +181,9 @@ func (h *PushHandler) NotifyNewMsg(_ context.Context, msgID int64, addr string) if item.Reaction != nil { return // FMSG-005: a reaction does not warrant a notification } + if item.purged || item.Trash != nil { + return // in the recipient's trash: they aren't told + } payload, err := json.Marshal(buildPushPayload(item, rootID, h.iconURL)) if err != nil { diff --git a/internal/handlers/reactions.go b/internal/handlers/reactions.go index 28bbde3..fae4c5c 100644 --- a/internal/handlers/reactions.go +++ b/internal/handlers/reactions.go @@ -326,7 +326,7 @@ func (h *MessageHandler) React(c *gin.Context) { return } - if draftHidden([]string{identity}, subject.From, subject.Time) { + if draftHidden([]string{identity}, subject.From, subject.Time) || purgedFor(ctx, h.q(), identity, msgID) { c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) return } diff --git a/internal/handlers/read_event_test.go b/internal/handlers/read_event_test.go index ae6ced3..f00e6c1 100644 --- a/internal/handlers/read_event_test.go +++ b/internal/handlers/read_event_test.go @@ -73,3 +73,40 @@ func TestMarkReadNotifies(t *testing.T) { t.Errorf("second read notified %q", n.Payload) } } + +// A msg_trash notification for a message deleted for good pushes only its id. +func TestHubRoute_MsgTrashPurged(t *testing.T) { + hub := &Hub{ + buildItem: func(_ context.Context, msgID int64, _ string) (*messageListItem, error) { + return &messageListItem{ID: msgID, purged: true}, nil + }, + registry: make(map[string]map[*wsClient]struct{}), + } + c := &wsClient{addr: "@bob@example.com", send: make(chan []byte, 4)} + hub.Register(c) + + hub.route(context.Background(), "msg_trash", 5, "@bob@example.com") + select { + case frame := <-c.send: + var env struct { + Type string `json:"type"` + Data map[string]any `json:"data"` + } + if err := json.Unmarshal(frame, &env); err != nil { + t.Fatal(err) + } + if env.Type != eventTrash || env.Data["id"] != float64(5) || env.Data["trash"] != trashPurged || len(env.Data) != 2 { + t.Errorf("frame %s", frame) + } + default: + t.Fatal("no frame pushed") + } + + // Other events about it are not pushed at all. + hub.route(context.Background(), "new_msg", 5, "@bob@example.com") + select { + case frame := <-c.send: + t.Errorf("pushed %s for a purged message", frame) + default: + } +} diff --git a/internal/handlers/references.go b/internal/handlers/references.go index 696683f..fb36b70 100644 --- a/internal/handlers/references.go +++ b/internal/handlers/references.go @@ -104,6 +104,11 @@ func (h *MessageHandler) resolveParent(c *gin.Context, in *messageInput) bool { c.JSON(403, gin.H{"error": "not a participant of this parent; added recipients must use their batch sha256"}) return false } + // A parent deleted for good no longer exists for the caller. + if purgedFor(ctx, h.q(), middleware.GetIdentity(c), id) { + c.JSON(400, gin.H{"error": "parent message not found"}) + return false + } if sent == nil || terminal { c.JSON(409, gin.H{"error": "parent is a draft or terminal message"}) return false diff --git a/internal/handlers/subaccounts_lifecycle_test.go b/internal/handlers/subaccounts_lifecycle_test.go index e014dc6..e328fcb 100644 --- a/internal/handlers/subaccounts_lifecycle_test.go +++ b/internal/handlers/subaccounts_lifecycle_test.go @@ -139,6 +139,18 @@ func newSubAccountTestStore(t *testing.T) *apiauth.Store { t.Fatal(err) } t.Cleanup(pool.Close) + // This service's dd.sql extends the fmsgd schema. + fmsgdDD := os.Getenv("FMSG_TEST_DD") + if fmsgdDD == "" { + fmsgdDD = "../../../fmsgd/dd.sql" + } + base, err := os.ReadFile(fmsgdDD) + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, string(base)); err != nil { + t.Fatal(err) + } dd, err := os.ReadFile("../../dd.sql") if err != nil { t.Fatal(err) diff --git a/internal/handlers/thread.go b/internal/handlers/thread.go index bf7833f..fe229b2 100644 --- a/internal/handlers/thread.go +++ b/internal/handlers/thread.go @@ -63,6 +63,7 @@ type threadMessage struct { Type string `json:"type,omitempty"` Size int `json:"size,omitempty"` MessageSHA256 string `json:"message_sha256,omitempty"` + Trash *string `json:"trash,omitempty"` // "message" or "thread" when in the caller's trash Body *threadBody `json:"body,omitempty"` Attachments []threadAttachment `json:"attachments,omitempty"` dataPath string @@ -277,6 +278,17 @@ func (h *MessageHandler) ThreadText(c *gin.Context) { return } ctx := c.Request.Context() + hasTrash, err := h.hasTrash(ctx, addrs[0]) + if err != nil { + log.Printf("thread text: trash: %v", err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + // A message deleted for good is not found, as for one that doesn't exist. + if hasTrash && purgedFor(ctx, h.q(), addrs[0], msgID) { + c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) + return + } rows, err := h.q().Query(ctx, ` WITH RECURSIVE chain AS ( @@ -291,9 +303,10 @@ func (h *MessageHandler) ThreadText(c *gin.Context) { (lower(c.from_addr) = ANY($2) OR (c.time_sent IS NOT NULL AND (EXISTS (SELECT 1 FROM msg_to t WHERE t.msg_id = c.id AND lower(t.addr) = ANY($2)) - OR EXISTS (SELECT 1 FROM msg_add_to a WHERE a.msg_id = c.id AND lower(a.addr) = ANY($2))))) AS readable + OR EXISTS (SELECT 1 FROM msg_add_to a WHERE a.msg_id = c.id AND lower(a.addr) = ANY($2))))) + AND (NOT $4 OR fmsg_api_trash_state($2[1], c.id) IS DISTINCT FROM 'purged') AS readable FROM chain c ORDER BY c.depth DESC`, - msgID, addrs, threadMaxHops, + msgID, addrs, threadMaxHops, hasTrash, ) if err != nil { log.Printf("thread text: walk %d: %v", msgID, err) @@ -376,6 +389,17 @@ func (h *MessageHandler) ThreadMessages(c *gin.Context) { return } ctx := c.Request.Context() + hasTrash, err := h.hasTrash(ctx, addrs[0]) + if err != nil { + log.Printf("thread messages: trash: %v", err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) + return + } + // A message deleted for good is not found, as for one that doesn't exist. + if hasTrash && purgedFor(ctx, h.q(), addrs[0], msgID) { + c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) + return + } 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) @@ -402,8 +426,9 @@ func (h *MessageHandler) ThreadMessages(c *gin.Context) { (lower(c.from_addr) = ANY($2) OR (c.time_sent IS NOT NULL AND (EXISTS (SELECT 1 FROM msg_to t WHERE t.msg_id = c.id AND lower(t.addr) = ANY($2)) - OR EXISTS (SELECT 1 FROM msg_add_to a WHERE a.msg_id = c.id AND lower(a.addr) = ANY($2))))) - FROM chain c ORDER BY c.depth DESC`, msgID, addrs, threadMaxHops) + OR EXISTS (SELECT 1 FROM msg_add_to a WHERE a.msg_id = c.id AND lower(a.addr) = ANY($2))))), + CASE WHEN $4 THEN fmsg_api_trash_state($2[1], c.id) END + FROM chain c ORDER BY c.depth DESC`, msgID, addrs, threadMaxHops, hasTrash) if err != nil { log.Printf("thread messages: walk %d: %v", msgID, err) c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) @@ -415,7 +440,7 @@ func (h *MessageHandler) ThreadMessages(c *gin.Context) { var hash *string if err = rows.Scan(&m.ID, &m.Version, &m.PID, &m.NoReply, &m.Important, &m.Deflate, &m.Terminal, &m.Time, &m.From, &m.Topic, &m.Type, &m.Size, &m.dataPath, - &hash, &m.PSHA256, &m.Visible); err != nil { + &hash, &m.PSHA256, &m.Visible, &m.Trash); err != nil { rows.Close() c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to retrieve thread"}) return @@ -424,6 +449,13 @@ func (h *MessageHandler) ThreadMessages(c *gin.Context) { m.MessageSHA256 = *hash m.SHA256 = hash } + // A message deleted for good is no longer visible to the caller. + if m.Trash != nil && *m.Trash == trashPurged { + m.Visible = false + } + if !m.Visible { + m.Trash = nil + } messages = append(messages, m) } rows.Close() diff --git a/internal/handlers/trash.go b/internal/handlers/trash.go new file mode 100644 index 0000000..fd27b01 --- /dev/null +++ b/internal/handlers/trash.go @@ -0,0 +1,520 @@ +package handlers + +import ( + "context" + "fmt" + "log" + "net/http" + "os" + "strings" + "time" + + "github.com/gin-gonic/gin" + "github.com/markmnl/fmsg-webapi/internal/middleware" +) + +// Trash. Each address has its own: deleting a sent message only hides it from +// that address, and replies keep arriving. A thread mark hides everything +// below the message it is on, including replies that arrive later. Deleting +// for good (purging) hides the message and everything that replies to it from +// that address for ever, gives back the storage it counted against that +// address, and removes it from the host once no local participant still +// holds it; a reply to it is then rejected as parent not found. State lives +// in fmsg_api_trash and is read through fmsg_api_trash_state (dd.sql). + +const ( + trashMessage = "message" // deleted itself + trashThread = "thread" // in a deleted thread + trashPurged = "purged" // deleted for good +) + +// trashStateQ is where msgID stands for addr: "", trashMessage, trashThread +// or trashPurged. An address that has never deleted anything costs one +// indexed lookup. +func trashStateQ(ctx context.Context, q queries, addr string, msgID int64) (string, error) { + var state *string + err := q.QueryRow(ctx, + `SELECT CASE WHEN EXISTS (SELECT 1 FROM fmsg_api_trash WHERE addr = $1) + THEN fmsg_api_trash_state($1, $2) END`, + strings.ToLower(addr), msgID).Scan(&state) + if err != nil || state == nil { + return "", err + } + return *state, nil +} + +// purgedFor reports whether msgID is deleted for good by addr. Lookup errors +// are logged and treated as not purged. +func purgedFor(ctx context.Context, q queries, addr string, msgID int64) bool { + state, err := trashStateQ(ctx, q, addr, msgID) + if err != nil { + log.Printf("trash state of %d for %s: %v", msgID, addr, err) + } + return state == trashPurged +} + +// hasTrash reports whether addr has deleted anything. +func (h *MessageHandler) hasTrash(ctx context.Context, addr string) (bool, error) { + var has bool + err := h.q().QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM fmsg_api_trash WHERE addr = $1)`, strings.ToLower(addr)).Scan(&has) + return has, err +} + +// trashListFilter is the list condition for a ?trash= mode on a row whose +// state is in the column state. Purged messages are never listed. +func trashListFilter(c *gin.Context) (string, bool) { + switch mode := c.DefaultQuery("trash", "exclude"); mode { + case "exclude": + return "t.state IS NULL", true + case "include": + return "t.state IS DISTINCT FROM 'purged'", true + case "only": + return "t.state IN ('message', 'thread')", true + default: + c.JSON(http.StatusBadRequest, gin.H{"error": "trash must be exclude, include or only"}) + return "", false + } +} + +// localDomains are the domains whose addresses hold their messages on this +// host. +func (h *MessageHandler) localDomains() []string { + if h.LocalDomain == "" { + return nil + } + return []string{strings.ToLower(h.LocalDomain)} +} + +// boolQuery reads a ?name=true flag. +func boolQuery(c *gin.Context, name string) bool { + v := strings.ToLower(c.Query(name)) + return v == "true" || v == "1" +} + +// participates reports whether addr sent msgID or is one of its recipients. +func (h *MessageHandler) participates(ctx context.Context, addr string, msgID int64) (bool, error) { + var ok bool + err := h.q().QueryRow(ctx, `SELECT + EXISTS (SELECT 1 FROM msg WHERE id = $1 AND lower(from_addr) = $2) + OR EXISTS (SELECT 1 FROM msg_to WHERE msg_id = $1 AND lower(addr) = $2) + OR EXISTS (SELECT 1 FROM msg_add_to WHERE msg_id = $1 AND lower(addr) = $2)`, + msgID, strings.ToLower(addr)).Scan(&ok) + return ok, err +} + +// trashSubtree lists msgID and the sent messages below it that addr takes +// part in, deepest first. +func (h *MessageHandler) trashSubtree(ctx context.Context, addr string, msgID int64) ([]int64, error) { + rows, err := h.q().Query(ctx, ` + WITH RECURSIVE sub(id, depth) AS ( + SELECT $1::bigint, 0 + UNION ALL + SELECT m.id, s.depth + 1 FROM msg m JOIN sub s ON m.pid = s.id WHERE s.depth < 1000 + ) + SELECT s.id FROM sub s JOIN msg m ON m.id = s.id + WHERE m.time_sent IS NOT NULL + AND (lower(m.from_addr) = $2 + OR EXISTS (SELECT 1 FROM msg_to t WHERE t.msg_id = s.id AND lower(t.addr) = $2) + OR EXISTS (SELECT 1 FROM msg_add_to a WHERE a.msg_id = s.id AND lower(a.addr) = $2)) + GROUP BY s.id ORDER BY max(s.depth) DESC, s.id DESC`, + msgID, strings.ToLower(addr)) + if err != nil { + return nil, err + } + defer rows.Close() + var ids []int64 + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids = append(ids, id) + } + return ids, rows.Err() +} + +// notifyTrash tells addr's connections that ids moved in or out of the trash. +// The notifications are sent when the transaction commits. +func (h *MessageHandler) notifyTrash(ctx context.Context, addr string, ids []int64) { + for _, id := range ids { + if _, err := h.q().Exec(ctx, `SELECT pg_notify('msg_trash', $1)`, fmt.Sprintf("%d,%s", id, addr)); err != nil { + log.Printf("trash %d: notify: %v", id, err) + } + } +} + +// trashSent handles DELETE /fmsg/:id for a sent message: it moves it (or with +// ?thread=true its thread) to the caller's trash, or with ?permanent=true +// deletes it for good. Runs in the request's transaction. +func (h *MessageHandler) trashSent(c *gin.Context, msgID int64) { + identity := strings.ToLower(middleware.GetIdentity(c)) + ctx := c.Request.Context() + + ok, err := h.participates(ctx, identity, msgID) + if err != nil { + log.Printf("trash %d: participation: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete message"}) + return + } + state, err := trashStateQ(ctx, h.q(), identity, msgID) + if err != nil { + log.Printf("trash %d: state: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete message"}) + return + } + if !ok || state == trashPurged { + c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) + return + } + + thread := boolQuery(c, "thread") + if boolQuery(c, "permanent") { + h.purge(c, identity, msgID, thread) + return + } + + now := float64(time.Now().UnixMicro()) / 1e6 + if _, err := h.q().Exec(ctx, ` + INSERT INTO fmsg_api_trash (addr, msg_id, thread, time_deleted) VALUES ($1, $2, $3, $4) + ON CONFLICT (addr, msg_id) DO UPDATE SET thread = fmsg_api_trash.thread OR EXCLUDED.thread`, + identity, msgID, thread, now); err != nil { + log.Printf("trash %d: mark: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete message"}) + return + } + ids := []int64{msgID} + if thread { + if ids, err = h.trashSubtree(ctx, identity, msgID); err != nil { + log.Printf("trash %d: thread: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete message"}) + return + } + } + h.notifyTrash(ctx, identity, ids) + if thread { + state = trashThread + } else if state, err = trashStateQ(ctx, h.q(), identity, msgID); err != nil { + log.Printf("trash %d: state: %v", msgID, err) + state = trashMessage + } + c.JSON(http.StatusOK, gin.H{"id": msgID, "trash": state}) +} + +// Restore handles POST /fmsg/:id/restore — takes a message (or with +// ?thread=true the thread below it) out of the caller's trash. Messages +// deleted for good can't be restored. +func (h *MessageHandler) Restore(c *gin.Context) { + identity := strings.ToLower(middleware.GetIdentity(c)) + msgID, ok := resolveID(c, h.q()) + if !ok { + return + } + ctx := c.Request.Context() + + ok, err := h.participates(ctx, identity, msgID) + if err != nil { + log.Printf("restore %d: participation: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to restore message"}) + return + } + if !ok || purgedFor(ctx, h.q(), identity, msgID) { + c.JSON(http.StatusNotFound, gin.H{"error": "message not found"}) + return + } + + ids := []int64{msgID} + if boolQuery(c, "thread") { + if ids, err = h.trashSubtree(ctx, identity, msgID); err != nil { + log.Printf("restore %d: thread: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to restore message"}) + return + } + } + if _, err := h.q().Exec(ctx, + `DELETE FROM fmsg_api_trash WHERE addr = $1 AND msg_id = ANY($2) AND time_purged IS NULL`, + identity, ids); err != nil { + log.Printf("restore %d: %v", msgID, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to restore message"}) + return + } + h.notifyTrash(ctx, identity, ids) + state, err := trashStateQ(ctx, h.q(), identity, msgID) + if err != nil { + log.Printf("restore %d: state: %v", msgID, err) + } + var trash *string + if state != "" { + trash = &state + } + c.JSON(http.StatusOK, gin.H{"id": msgID, "trash": trash}) +} + +// usageRelease is storage to give back to fmsgid once the purge commits. +type usageRelease struct { + send bool + size int64 +} + +// purge deletes msgID and everything below it for good for identity (with +// thread, msgID is the thread's top as the caller sees it). It gives back +// what each newly purged message counted against identity, and removes from +// the host every purged message no local participant still holds. +func (h *MessageHandler) purge(c *gin.Context, identity string, msgID int64, thread bool) { + ctx := c.Request.Context() + fail := func(what string, err error) { + log.Printf("purge %d: %s: %v", msgID, what, err) + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete message"}) + } + + ids, err := h.trashSubtree(ctx, identity, msgID) + if err != nil { + fail("subtree", err) + return + } + + // What each message newly deleted for good counted against identity. + var releases []usageRelease + var purged int + now := float64(time.Now().UnixMicro()) / 1e6 + for _, id := range ids { + var already bool + if err := h.q().QueryRow(ctx, + `SELECT EXISTS (SELECT 1 FROM fmsg_api_trash WHERE addr = $1 AND msg_id = $2 AND time_purged IS NOT NULL)`, + identity, id).Scan(&already); err != nil { + fail("purged state", err) + return + } + if already { + continue + } + if _, err := h.q().Exec(ctx, ` + INSERT INTO fmsg_api_trash (addr, msg_id, thread, time_deleted, time_purged) VALUES ($1, $2, $3, $4, $4) + ON CONFLICT (addr, msg_id) DO UPDATE SET time_purged = EXCLUDED.time_purged, + thread = fmsg_api_trash.thread OR EXCLUDED.thread`, + identity, id, thread && id == msgID, now); err != nil { + fail("mark", err) + return + } + purged++ + rel, err := h.usageOf(ctx, identity, id) + if err != nil { + fail("usage", err) + return + } + releases = append(releases, rel...) + } + + // Remove what nobody here holds any more, deepest first, so a reply + // is gone before the message it replies to. + var removed int + var files []string + for _, id := range ids { + gone, paths, err := h.removeIfUnheld(ctx, id) + if err != nil { + fail("remove", err) + return + } + if gone { + removed++ + files = append(files, paths...) + } + } + h.notifyTrash(ctx, identity, ids) + + for _, p := range files { + path := p + afterCommit(c, func() { + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + log.Printf("purge: remove %s: %v", path, err) + } + }) + } + if h.IDURL != "" && len(releases) > 0 { + idURL := h.IDURL + afterCommit(c, func() { + for _, r := range releases { + var err error + if r.send { + err = middleware.ReleaseFmsgIDSend(idURL, identity, now, r.size) + } else { + err = middleware.ReleaseFmsgIDRecv(idURL, identity, now, r.size) + } + if err != nil { + log.Printf("release usage of %s: %v", identity, err) + } + } + }) + } + c.JSON(http.StatusOK, gin.H{"id": msgID, "trash": trashPurged, "purged": purged, "removed": removed}) +} + +// usageOf is what msgID counted against addr's quota: its size as sent if +// addr sent it, and as received for each delivery to addr. The size is the +// one recorded: messageStoredSize for a message sent from this host, and the +// expanded size fmsgd records for one it received. +func (h *MessageHandler) usageOf(ctx context.Context, addr string, msgID int64) ([]usageRelease, error) { + var from string + var sent bool + var deliveries int + if err := h.q().QueryRow(ctx, `SELECT m.from_addr, m.time_sent IS NOT NULL, + (SELECT count(*) FROM msg_to WHERE msg_id = m.id AND lower(addr) = $2 AND time_delivered IS NOT NULL) + + (SELECT count(*) FROM msg_add_to WHERE msg_id = m.id AND lower(addr) = $2 AND time_delivered IS NOT NULL) + FROM msg m WHERE m.id = $1`, msgID, addr).Scan(&from, &sent, &deliveries); err != nil { + return nil, err + } + isSender := sent && strings.EqualFold(from, addr) + if !isSender && deliveries == 0 { + return nil, nil + } + size, err := h.recordedSize(ctx, msgID, from) + if err != nil { + return nil, err + } + var out []usageRelease + if isSender { + out = append(out, usageRelease{send: true, size: size}) + } + for i := 0; i < deliveries; i++ { + out = append(out, usageRelease{size: size}) + } + return out, nil +} + +// recordedSize is the size a message was recorded at for quotas. +func (h *MessageHandler) recordedSize(ctx context.Context, msgID int64, from string) (int64, error) { + _, domain := parseAddr(from) + for _, d := range h.localDomains() { + if strings.EqualFold(domain, d) { + return h.messageStoredSize(ctx, msgID) + } + } + // Received by fmsgd, which records the expanded size: what is stored + // on disk for a compressed part, the wire size otherwise. + var size int64 + var deflate bool + var body string + if err := h.q().QueryRow(ctx, `SELECT size, is_deflate, filepath FROM msg WHERE id = $1`, msgID).Scan(&size, &deflate, &body); err != nil { + return 0, err + } + if deflate { + size = h.fileSize(body, size) + } + rows, err := h.q().Query(ctx, `SELECT flags, filesize, filepath FROM msg_attachment WHERE msg_id = $1`, msgID) + if err != nil { + return 0, err + } + defer rows.Close() + for rows.Next() { + var flags int + var fsize int64 + var path string + if err := rows.Scan(&flags, &fsize, &path); err != nil { + return 0, err + } + if flags&(1<<1) != 0 { + fsize = h.fileSize(path, fsize) + } + size += fsize + } + return size, rows.Err() +} + +// fileSize is the size of a stored file, or fallback if it can't be read. +func (h *MessageHandler) fileSize(path string, fallback int64) int64 { + clean, ok := safeDataPath(path, h.DataDir) + if !ok { + return fallback + } + if fi, err := os.Stat(clean); err == nil { + return fi.Size() + } + return fallback +} + +// removeIfUnheld removes msgID from the host when no local participant holds +// it any more (none sent or accepted it without deleting it for good) and no +// delivery or add-to notification of it is still pending. It returns whether +// it was removed and the stored files to delete once the transaction commits. +func (h *MessageHandler) removeIfUnheld(ctx context.Context, msgID int64) (bool, []string, error) { + var held, pending bool + err := h.q().QueryRow(ctx, `SELECT + EXISTS ( + SELECT 1 FROM ( + SELECT from_addr AS a FROM msg WHERE id = $1 AND time_sent IS NOT NULL + UNION ALL SELECT addr FROM msg_to WHERE msg_id = $1 AND time_delivered IS NOT NULL + UNION ALL SELECT addr FROM msg_add_to WHERE msg_id = $1 AND time_delivered IS NOT NULL + ) p + WHERE split_part(lower(p.a), '@', 3) = ANY($2) + AND NOT EXISTS (SELECT 1 FROM fmsg_api_trash t + WHERE t.addr = lower(p.a) AND t.msg_id = $1 AND t.time_purged IS NOT NULL)), + EXISTS (SELECT 1 FROM msg_to WHERE msg_id = $1 AND time_delivered IS NULL AND (response_code IS NULL OR response_code = -1)) + OR EXISTS (SELECT 1 FROM msg_add_to WHERE msg_id = $1 AND time_delivered IS NULL AND (response_code IS NULL OR response_code = -1)) + OR EXISTS (SELECT 1 FROM msg_add_to_notify n JOIN msg_add_to_batch b ON b.id = n.batch_id + WHERE b.msg_id = $1 AND n.time_notified IS NULL)`, + msgID, h.localDomains()).Scan(&held, &pending) + if err != nil || held || pending { + return false, nil, err + } + + var paths []string + var body string + if err := h.q().QueryRow(ctx, `SELECT filepath FROM msg WHERE id = $1`, msgID).Scan(&body); err != nil { + return false, nil, err + } + paths = append(paths, body) + rows, err := h.q().Query(ctx, `SELECT filepath FROM msg_attachment WHERE msg_id = $1`, msgID) + if err != nil { + return false, nil, err + } + for rows.Next() { + var p string + if err := rows.Scan(&p); err != nil { + rows.Close() + return false, nil, err + } + paths = append(paths, p) + } + rows.Close() + + // fmsgd keeps sent messages immutable unless the transaction says it is + // purging (fmsgd dd.sql, protect_msg_parts). + for _, q := range []string{ + `SET LOCAL fmsg.purge = 'on'`, + `UPDATE msg SET pid = NULL WHERE pid = $1`, + `DELETE FROM msg_add_to_notify WHERE batch_id IN (SELECT id FROM msg_add_to_batch WHERE msg_id = $1)`, + `DELETE FROM msg_add_to WHERE msg_id = $1`, + `DELETE FROM msg_add_to_batch WHERE msg_id = $1`, + `DELETE FROM msg_to WHERE msg_id = $1`, + `DELETE FROM msg_attachment WHERE msg_id = $1`, + `DELETE FROM msg WHERE id = $1`, + } { + var err error + if strings.Contains(q, "$1") { + _, err = h.q().Exec(ctx, q, msgID) + } else { + _, err = h.q().Exec(ctx, q) + } + if err != nil { + return false, nil, fmt.Errorf("%s: %w", q, err) + } + } + + // Only delete files inside the data directory that no other message + // still uses. + var files []string + for _, p := range paths { + clean, ok := safeDataPath(p, h.DataDir) + if !ok { + continue + } + var shared bool + if err := h.q().QueryRow(ctx, `SELECT EXISTS (SELECT 1 FROM msg WHERE filepath = $1) + OR EXISTS (SELECT 1 FROM msg_attachment WHERE filepath = $1)`, p).Scan(&shared); err != nil { + return false, nil, err + } + if !shared { + files = append(files, clean) + } + } + return true, files, nil +} diff --git a/internal/handlers/trash_integration_test.go b/internal/handlers/trash_integration_test.go new file mode 100644 index 0000000..3399c80 --- /dev/null +++ b/internal/handlers/trash_integration_test.go @@ -0,0 +1,311 @@ +package handlers + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "os" + "sync" + "testing" + "time" +) + +const ( + alice = "@alice@example.com" + bob = "@bob@example.com" + carol = "@carol@example.com" +) + +// usageLog records the usage fmsgid is told about. +type usageLog struct { + mu sync.Mutex + posts []string // "
" +} + +func (u *usageLog) handler() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPost { + var tx struct { + Address string `json:"address"` + TS float64 `json:"ts"` + Size int64 `json:"size"` + } + body, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(body, &tx) + u.mu.Lock() + u.posts = append(u.posts, fmt.Sprintf("%s %s %d", r.URL.Path, tx.Address, tx.Size)) + u.mu.Unlock() + return + } + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"acceptingNew":true}`) + }) +} + +// releases lists the release posts made so far. +func (u *usageLog) releases() []string { + u.mu.Lock() + defer u.mu.Unlock() + var out []string + for _, p := range u.posts { + if len(p) > len("/fmsgid/send/release") && (p[:len("/fmsgid/send/release")] == "/fmsgid/send/release" || p[:len("/fmsgid/recv/release")] == "/fmsgid/recv/release") { + out = append(out, p) + } + } + return out +} + +// sendAs sends a new message from `from` to `to` (replying to pid when set) +// and returns its id. +func (a *finalizationAPI) sendAs(t *testing.T, from string, to []string, body string, pid any) string { + t.Helper() + v := jsonResponse(t, a.request(from, "POST", "/fmsg", map[string]any{ + "version": 1, "from": from, "to": to, "type": "text/plain;charset=UTF-8", "data": body, "pid": pid, + }), 201) + id := fmt.Sprintf("%.0f", v["id"]) + jsonResponse(t, a.request(from, "POST", "/fmsg/"+id+"/send", map[string]any{}), 200) + return id +} + +// listed returns the trash state of each message in identity's inbox for the +// trash mode ("" for none). +func (a *finalizationAPI) listed(t *testing.T, identity, mode string) map[string]any { + t.Helper() + path := "/fmsg?limit=100" + if mode != "" { + path += "&trash=" + mode + } + w := a.request(identity, "GET", path, nil) + if w.Code != 200 { + t.Fatalf("list %s: %d %s", mode, w.Code, w.Body) + } + var items []struct { + ID json.Number `json:"id"` + Trash any `json:"trash"` + } + if err := json.Unmarshal(w.Body.Bytes(), &items); err != nil { + t.Fatal(err) + } + out := map[string]any{} + for _, it := range items { + out[it.ID.String()] = it.Trash + } + return out +} + +func wantListed(t *testing.T, got map[string]any, id string, want any) { + t.Helper() + state, ok := got[id] + if want == "absent" { + if ok { + t.Errorf("message %s listed (trash %v), want absent", id, state) + } + return + } + if !ok { + t.Errorf("message %s not listed, want trash %v", id, want) + return + } + if state != want { + t.Errorf("message %s trash = %v, want %v", id, state, want) + } +} + +// Deleting moves a message, or a whole thread with its later replies, into the +// deleter's trash only, and restoring brings it back. +func TestTrashAndRestore(t *testing.T) { + a := newFinalizationAPI(t) + root := a.sendAs(t, alice, []string{bob, alice}, "root", nil) + reply := a.sendAs(t, alice, []string{bob, alice}, "reply", mustID(t, root)) + + if w := a.request(bob, "GET", "/fmsg?trash=bogus", nil); w.Code != 400 { + t.Errorf("bad trash mode: HTTP %d, want 400", w.Code) + } + if w := a.request(carol, "DELETE", "/fmsg/"+root, nil); w.Code != 404 { + t.Errorf("non-participant delete: HTTP %d, want 404", w.Code) + } + + // One message. + v := jsonResponse(t, a.request(bob, "DELETE", "/fmsg/"+root, nil), 200) + if v["trash"] != "message" { + t.Errorf("delete response trash = %v", v["trash"]) + } + wantListed(t, a.listed(t, bob, ""), root, "absent") + wantListed(t, a.listed(t, bob, ""), reply, nil) + wantListed(t, a.listed(t, bob, "include"), root, "message") + only := a.listed(t, bob, "only") + wantListed(t, only, root, "message") + wantListed(t, only, reply, "absent") + if got := jsonResponse(t, a.request(bob, "GET", "/fmsg/"+root, nil), 200)["trash"]; got != "message" { + t.Errorf("get trash = %v", got) + } + wantListed(t, a.listed(t, alice, ""), root, nil) // alice's inbox is her own + + // The whole thread, including a reply that arrives afterwards. + jsonResponse(t, a.request(bob, "DELETE", "/fmsg/"+root+"?thread=true", nil), 200) + later := a.sendAs(t, alice, []string{bob, alice}, "later", mustID(t, reply)) + def := a.listed(t, bob, "") + for _, id := range []string{root, reply, later} { + wantListed(t, def, id, "absent") + } + inc := a.listed(t, bob, "include") + wantListed(t, inc, root, "thread") + wantListed(t, inc, later, "thread") + item, err := a.h.messageItemFor(context.Background(), mustID(t, later), bob) + if err != nil || item.Trash == nil || *item.Trash != "thread" { + t.Fatalf("pushed item trash = %v, %v; a reply in a deleted thread must not be pushed", item, err) + } + // It still shows as unread there. + if item.Read { + t.Error("later reply should be unread") + } + + // Restore the thread. + jsonResponse(t, a.request(bob, "POST", "/fmsg/"+root+"/restore?thread=true", map[string]any{}), 200) + def = a.listed(t, bob, "") + for _, id := range []string{root, reply, later} { + wantListed(t, def, id, nil) + } + + // Drafts are still deleted outright by their owner. + draft := a.draft(t, alice, "unsent", nil) + if w := a.request(alice, "DELETE", "/fmsg/"+draft, nil); w.Code != 204 { + t.Errorf("draft delete: HTTP %d, want 204", w.Code) + } +} + +// Deleting for good hides a message from the deleter everywhere, gives back +// its usage, and removes it from the host once nobody here holds it. +func TestPurge(t *testing.T) { + usage := &usageLog{} + a := newFinalizationAPIWithID(t, usage.handler()) + ctx := context.Background() + root := a.sendAs(t, alice, []string{bob, alice}, "root", nil) + reply := a.sendAs(t, bob, []string{alice, bob, carol}, "reply", mustID(t, root)) + var body, hash string + if err := a.pool.QueryRow(ctx, `SELECT filepath, encode(sha256,'hex') FROM msg WHERE id = $1`, mustID(t, root)).Scan(&body, &hash); err != nil { + t.Fatal(err) + } + size, err := a.h.messageStoredSize(ctx, mustID(t, root)) + if err != nil { + t.Fatal(err) + } + + // Bob deletes the root for good: alice still holds it. + v := jsonResponse(t, a.request(bob, "DELETE", "/fmsg/"+root+"?permanent=true", nil), 200) + if v["purged"] != float64(2) || v["removed"] != float64(0) { // the root and bob's reply below it + t.Fatalf("bob purge = %v", v) + } + for _, r := range []struct{ method, path string }{ + {"GET", "/fmsg/" + root}, + {"GET", "/fmsg/" + root + "/data"}, + {"GET", "/fmsg/" + root + "/thread"}, + {"GET", "/fmsg/" + root + "/thread/messages"}, + {"POST", "/fmsg/" + root + "/read"}, + {"POST", "/fmsg/" + root + "/restore"}, + {"DELETE", "/fmsg/" + root}, + } { + if w := a.request(bob, r.method, r.path, map[string]any{}); w.Code != 404 { + t.Errorf("bob %s %s after purge: HTTP %d, want 404", r.method, r.path, w.Code) + } + } + wantListed(t, a.listed(t, bob, "include"), root, "absent") + if w := a.request(bob, "POST", "/fmsg", map[string]any{"version": 1, "from": bob, "to": []string{alice}, "type": "text/plain;charset=UTF-8", "data": "re", "pid": mustID(t, root)}); w.Code != 400 { + t.Errorf("reply to a purged parent: HTTP %d, want 400: %s", w.Code, w.Body) + } + if w := a.request(alice, "GET", "/fmsg/"+root, nil); w.Code != 200 { + t.Errorf("alice get after bob's purge: HTTP %d, want 200", w.Code) + } + wantRelease(t, usage, fmt.Sprintf("/fmsgid/recv/release %s %d", bob, size)) + + // Alice deletes the root for good: nobody holds it now. Carol still + // holds the reply, which stays, no longer linked to it. + v = jsonResponse(t, a.request(alice, "DELETE", "/fmsg/"+root+"?permanent=true", nil), 200) + if v["removed"] != float64(1) { + t.Fatalf("alice purge = %v", v) + } + var n int + if err := a.pool.QueryRow(ctx, `SELECT count(*) FROM msg WHERE encode(sha256,'hex') = $1`, hash).Scan(&n); err != nil || n != 0 { + t.Fatalf("root still stored: %d, %v", n, err) + } + if _, err := os.Stat(body); !os.IsNotExist(err) { + t.Errorf("root body still on disk: %v", err) + } + var pid *int64 + if err := a.pool.QueryRow(ctx, `SELECT pid FROM msg WHERE id = $1`, mustID(t, reply)).Scan(&pid); err != nil || pid != nil { + t.Fatalf("reply after its parent was removed: pid %v, %v", pid, err) + } + if w := a.request(carol, "GET", "/fmsg/"+reply, nil); w.Code != 200 { + t.Errorf("carol's reply: HTTP %d, want 200", w.Code) + } + // Alice sent and received the root: both come back to her. + wantRelease(t, usage, fmt.Sprintf("/fmsgid/send/release %s %d", alice, size)) + wantRelease(t, usage, fmt.Sprintf("/fmsgid/recv/release %s %d", alice, size)) + + // Deleting the root for good took the reply below it with it, once. + replySize, err := a.h.messageStoredSize(ctx, mustID(t, reply)) + if err != nil { + t.Fatal(err) + } + wantRelease(t, usage, fmt.Sprintf("/fmsgid/recv/release %s %d", alice, replySize)) + before := len(usage.releases()) + if w := a.request(alice, "DELETE", "/fmsg/"+reply+"?permanent=true", nil); w.Code != 404 { + t.Errorf("purge twice: HTTP %d, want 404", w.Code) + } + if w := a.request(alice, "GET", "/fmsg/"+reply, nil); w.Code != 404 { + t.Errorf("alice get reply below her purged root: HTTP %d, want 404", w.Code) + } + if got := len(usage.releases()) - before; got != 0 { + t.Errorf("purging again released %d more", got) + } +} + +// A message still being delivered elsewhere is not removed. +func TestPurgeKeepsPendingDelivery(t *testing.T) { + a := newFinalizationAPI(t) + id := a.sendAs(t, alice, []string{"@dave@remote.example"}, "out", nil) + v := jsonResponse(t, a.request(alice, "DELETE", "/fmsg/"+id+"?permanent=true", nil), 200) + if v["removed"] != float64(0) { + t.Fatalf("pending delivery removed: %v", v) + } + if w := a.request(alice, "GET", "/fmsg/"+id, nil); w.Code != 404 { + t.Errorf("sender get after purge: HTTP %d, want 404", w.Code) + } +} + +func wantRelease(t *testing.T, u *usageLog, want string) { + t.Helper() + for _, p := range u.releases() { + if p == want { + return + } + } + t.Errorf("no release %q in %v", want, u.releases()) +} + +// Moving a message to the trash notifies msg_trash for the deleter. +func TestTrashNotifies(t *testing.T) { + a := newFinalizationAPI(t) + ctx := context.Background() + id := a.sendAs(t, alice, []string{bob, alice}, "hello", nil) + conn, err := a.pool.Acquire(ctx) + if err != nil { + t.Fatal(err) + } + defer conn.Release() + if _, err := conn.Exec(ctx, "LISTEN msg_trash"); err != nil { + t.Fatal(err) + } + jsonResponse(t, a.request(bob, "DELETE", "/fmsg/"+id, nil), 200) + wait, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() + n, err := conn.Conn().WaitForNotification(wait) + if err != nil { + t.Fatalf("no msg_trash notification: %v", err) + } + if want := id + "," + bob; n.Payload != want { + t.Errorf("payload %q, want %q", n.Payload, want) + } +} diff --git a/internal/middleware/fmsgid_usage.go b/internal/middleware/fmsgid_usage.go index 3e89312..27a2832 100644 --- a/internal/middleware/fmsgid_usage.go +++ b/internal/middleware/fmsgid_usage.go @@ -86,6 +86,20 @@ func RecordFmsgIDSend(idURL, addr string, ts float64, size int64) error { return postFmsgIDUsage(idURL, "send", addr, ts, size) } +// ReleaseFmsgIDSend gives back to addr the storage of a message it sent that +// is deleted for good: size bytes come off its sent size total, not off its +// daily usage. +func ReleaseFmsgIDSend(idURL, addr string, ts float64, size int64) error { + return postFmsgIDUsage(idURL, "send/release", addr, ts, size) +} + +// ReleaseFmsgIDRecv gives back to addr the storage of a message it received +// that is deleted for good: size bytes come off its received size total, not +// off its daily usage. +func ReleaseFmsgIDRecv(idURL, addr string, ts float64, size int64) error { + return postFmsgIDUsage(idURL, "recv/release", addr, ts, size) +} + func postFmsgIDUsage(idURL, part, addr string, ts float64, size int64) error { payload, err := json.Marshal(map[string]any{"address": addr, "ts": ts, "size": size}) if err != nil { diff --git a/internal/middleware/oauth.go b/internal/middleware/oauth.go index 5f3dc77..0079901 100644 --- a/internal/middleware/oauth.go +++ b/internal/middleware/oauth.go @@ -41,6 +41,7 @@ var oauthRouteScopes = map[string]string{ "DELETE /fmsg/:id": ScopeWrite, "POST /fmsg/:id/send": ScopeWrite, "POST /fmsg/:id/read": ScopeWrite, + "POST /fmsg/:id/restore": ScopeWrite, "POST /fmsg/:id/add-to": ScopeWrite, "POST /fmsg/:id/react": ScopeWrite, "POST /fmsg/:id/attach": ScopeWrite, diff --git a/internal/models/models.go b/internal/models/models.go index d77ee13..6660e2d 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -52,6 +52,7 @@ type Message struct { Attachments []Attachment `json:"attachments"` Reaction *string `json:"reaction"` // FMSG-005: the emoji this message carries when it is a reaction ("" clears); nil otherwise Reactions []Reaction `json:"reactions"` // FMSG-005: effective reactions on this message + Trash *string `json:"trash"` // "message" or "thread" when in the caller's trash, null otherwise } // Reaction is one emoji currently reacted on a message and the participants From f573a68de38a4ae256d85b80987aac762a67c5c1 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Thu, 22 Oct 2026 10:50:37 +0800 Subject: [PATCH 2/2] Use fmsgd with explicit purge Co-Authored-By: Claude Opus 5.5 --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index c854d2a..f54c572 100644 --- a/go.mod +++ b/go.mod @@ -10,7 +10,7 @@ require ( github.com/gorilla/websocket v1.5.3 github.com/jackc/pgx/v5 v5.8.0 github.com/joho/godotenv v1.5.1 - github.com/markmnl/fmsgd v0.6.1-0.20261010085601-c99adfef703e + github.com/markmnl/fmsgd v0.6.1-0.20261022024806-f5515e09b914 golang.org/x/sync v0.20.0 ) diff --git a/go.sum b/go.sum index 8274906..72e7555 100644 --- a/go.sum +++ b/go.sum @@ -60,8 +60,8 @@ github.com/leodido/go-urn v1.4.0 h1:WT9HwE9SGECu3lg4d/dIA+jxlljEa1/ffXKmRjqdmIQ= github.com/leodido/go-urn v1.4.0/go.mod h1:bvxc+MVxLKB4z00jd1z+Dvzr47oO32F/QSNjSBOlFxI= github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= -github.com/markmnl/fmsgd v0.6.1-0.20261010085601-c99adfef703e h1:VQf6zdJ1Hqd4+GhWnHgOwIGd64Z0j3OMJ9Z3046Bm0Y= -github.com/markmnl/fmsgd v0.6.1-0.20261010085601-c99adfef703e/go.mod h1:CiTt3R2Fw3zhEV8SDB7Ml1efdmYom5C/QOBT7GafEAM= +github.com/markmnl/fmsgd v0.6.1-0.20261022024806-f5515e09b914 h1:qSAvLtudh4oEwCOcgl89DWqm1loARku7ZMW9DjicKc0= +github.com/markmnl/fmsgd v0.6.1-0.20261022024806-f5515e09b914/go.mod h1:CiTt3R2Fw3zhEV8SDB7Ml1efdmYom5C/QOBT7GafEAM= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=