From 19a8ec95fae58bacde24e1bf5cadec2754dcdc82 Mon Sep 17 00:00:00 2001 From: Nixevol Date: Wed, 30 Sep 2026 08:06:55 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E7=AB=AF=E5=81=9C?= =?UTF-8?q?=E7=94=A8=E4=B8=8E=E5=88=A0=E9=99=A4=E7=9A=84=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E4=BD=9C=E5=BA=9F=E4=B8=8E=E7=BE=A4=E4=B8=BB=E8=BD=AC=E8=AE=A9?= =?UTF-8?q?=E7=BA=A7=E8=81=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/DEVIATIONS.md | 23 + internal/admin/endpoints.go | 37 +- internal/admin/endpoints_db.go | 33 +- internal/admin/handler.go | 5 + internal/app/identity/app.go | 6 +- internal/app/identity/lifecycle.go | 664 ++++++++++++++++++++++++ internal/app/identity/lifecycle_test.go | 386 ++++++++++++++ internal/app/identity/self.go | 5 - 8 files changed, 1143 insertions(+), 16 deletions(-) create mode 100644 internal/app/identity/lifecycle.go create mode 100644 internal/app/identity/lifecycle_test.go diff --git a/docs/DEVIATIONS.md b/docs/DEVIATIONS.md index 00855a6..c06cfcf 100644 --- a/docs/DEVIATIONS.md +++ b/docs/DEVIATIONS.md @@ -559,6 +559,29 @@ - 备选方案：测试里注入在线连接表使默认不保留也走 pending（与「离线不保留」场景重复覆盖）。 - 影响：仅测试期望；产品行为不变。 +### I5 2026-09-30 + +1. **停用/删除级联在 identity 包内完成，admin 可选注入** + - 原条款：DEVELOPMENT 7.6 / PRD F01；TASKS I5「提供给 A 线调用」。 + - 实际做法：`identity.App.Disable`/`Enable`/`Delete` 在一个写操作里完成启停、清令牌、作废消息/投递、发 `revoked`（可选 Downlink）、退群/转让群主/解散、清授权/回执/防重/发出记录。`admin.Deps.Identity` 非空时，`setEndpointEnabled`/`deleteEndpointBasic`（及 PATCH enabled）委托上述方法；为空时保留 A2 仅改库行为。未改 `cmd/nixmsg`、未改消息上行分发。 + - 原因：隔离要求不改接线；A 线已有路由，只接级联。 + - 备选方案：在 admin 内复制级联 SQL；或强制 Identity 必填。 + - 影响：生产须在挂载 admin 时注入 `identity.App`（及 KickEndpoint/ConnControl/Downlink），否则停用/删除仍无完整作废。 + +2. **消息级作废回执 state 用 `rejected`** + - 原条款：DEVELOPMENT 6.4 消息级作废写 `endpoint_id` 空、`state=rejected`；I4 解散群对 scheduled 曾写 `completed`。 + - 实际做法：I5 对「发给停用/删除端的 scheduled 单聊」及删除时解散群的 scheduled，回执 `state=rejected`，原因分别为 `endpoint_*` / `group_dissolved`。 + - 原因：与 6.4 字面一致。 + - 备选方案：与 I4 一样写 `completed`。 + - 影响：后台/SDK 若按 state 过滤回执需同时认 rejected。 + +3. **删除时群主转让按 `joined_at` 最早，并列按编号** + - 原条款：转给最早加入的其他成员。 + - 实际做法：`ORDER BY joined_at ASC, endpoint_id ASC LIMIT 1`。 + - 原因：同时加入时需稳定次序。 + - 备选方案：仅按 joined_at。 + - 影响：同毫秒加入时编号小者优先。 + ## 后台接口 A ### A1 2026-09-30 diff --git a/internal/admin/endpoints.go b/internal/admin/endpoints.go index 371b3a0..fe874b8 100644 --- a/internal/admin/endpoints.go +++ b/internal/admin/endpoints.go @@ -320,18 +320,39 @@ func (h *Handler) handleEndpointPatch(w http.ResponseWriter, r *http.Request) { return } - wasEnabled, err := h.patchEndpoint(r.Context(), id, req.Name, req.Remark, req.DefaultDelaySeconds, req.Enabled) - if err != nil { - if errors.Is(err, sql.ErrNoRows) { + hasMeta := req.Name != nil || req.Remark != nil || req.DefaultDelaySeconds != nil + var wasEnabled bool + var err error + if hasMeta || (req.Enabled != nil && h.identity == nil) { + wasEnabled, err = h.patchEndpoint(r.Context(), id, req.Name, req.Remark, req.DefaultDelaySeconds, req.Enabled) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + h.audit(actorString(p), "endpoint_patch", id, "not_found", ip) + httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在") + return + } + h.audit(actorString(p), "endpoint_patch", id, "error", ip) + httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误") + return + } + } + if req.Enabled != nil && h.identity != nil { + var found bool + found, err = h.setEndpointEnabled(r.Context(), id, *req.Enabled) + if err != nil { + h.audit(actorString(p), "endpoint_patch", id, "error", ip) + httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误") + return + } + if !found { h.audit(actorString(p), "endpoint_patch", id, "not_found", ip) httpx.WriteError(w, http.StatusNotFound, "not_found", "端不存在") return } - h.audit(actorString(p), "endpoint_patch", id, "error", ip) - httpx.WriteError(w, http.StatusInternalServerError, "internal", "内部错误") - return - } - if req.Enabled != nil && !*req.Enabled && wasEnabled { + if !*req.Enabled { + _, _ = h.kickEndpoint(r.Context(), id) + } + } else if req.Enabled != nil && !*req.Enabled && wasEnabled { _, _ = h.kickEndpoint(r.Context(), id) } row, err := h.getEndpoint(r.Context(), id) diff --git a/internal/admin/endpoints_db.go b/internal/admin/endpoints_db.go index 2903306..a32c16b 100644 --- a/internal/admin/endpoints_db.go +++ b/internal/admin/endpoints_db.go @@ -308,6 +308,21 @@ func (h *Handler) patchEndpoint(ctx context.Context, id string, name, remark *st } func (h *Handler) setEndpointEnabled(ctx context.Context, id string, enabled bool) (found bool, err error) { + if h.identity != nil { + var opErr error + if enabled { + opErr = h.identity.Enable(ctx, id) + } else { + opErr = h.identity.Disable(ctx, id) + } + if opErr != nil { + if isEndpointNotFound(opErr) { + return false, nil + } + return false, opErr + } + return true, nil + } err = h.db.Queue.Do(ctx, func(tx *sql.Tx) error { v := 0 if enabled { @@ -331,9 +346,18 @@ func (h *Handler) setEndpointEnabled(ctx context.Context, id string, enabled boo return found, err } -// deleteEndpointBasic 删除端行并清掉与之相关的 talk_grants。 -// 作废消息、群主转让等完整级联留给 I5。 +// deleteEndpointBasic 删除端；注入 Identity 时走 I5 完整级联。 func (h *Handler) deleteEndpointBasic(ctx context.Context, id string) (found bool, err error) { + if h.identity != nil { + opErr := h.identity.Delete(ctx, id) + if opErr != nil { + if isEndpointNotFound(opErr) { + return false, nil + } + return false, opErr + } + return true, nil + } err = h.db.Queue.Do(ctx, func(tx *sql.Tx) error { if _, e := tx.ExecContext(ctx, `DELETE FROM talk_grants WHERE sender_id = ? OR target_id = ?`, id, id); e != nil { return e @@ -349,6 +373,11 @@ func (h *Handler) deleteEndpointBasic(ctx context.Context, id string) (found boo return found, err } +func isEndpointNotFound(err error) bool { + var pe *protocol.Error + return errors.As(err, &pe) && pe.Code == protocol.CodeNotFound +} + func (h *Handler) resetLoginPassword(ctx context.Context, id, loginHash string) (bool, error) { var found bool err := h.db.Queue.Do(ctx, func(tx *sql.Tx) error { diff --git a/internal/admin/handler.go b/internal/admin/handler.go index e49ed71..445399d 100644 --- a/internal/admin/handler.go +++ b/internal/admin/handler.go @@ -7,6 +7,7 @@ import ( "sync" "time" + "git.asio.asia/nixevol/NixMsg/internal/app/identity" "git.asio.asia/nixevol/NixMsg/internal/auth" "git.asio.asia/nixevol/NixMsg/internal/store" ) @@ -38,6 +39,8 @@ type Deps struct { SecureCookies bool // KickEndpoint 踢下线钩子（只断开连接）；nil 时踢线为 no-op。 KickEndpoint EndpointKickFunc + // Identity 端停用/启用/删除级联（I5）；nil 时回退为仅改 enabled/删行。 + Identity identity.Service } // Handler 是可挂载的管理接口（路由前缀 /api/admin/）。 @@ -51,6 +54,7 @@ type Handler struct { ttl time.Duration forceSec bool kick EndpointKickFunc + identity identity.Service mux *http.ServeMux @@ -80,6 +84,7 @@ func New(d Deps) *Handler { ttl: ttl, forceSec: d.SecureCookies, kick: d.KickEndpoint, + identity: d.Identity, mux: http.NewServeMux(), lastUsed: make(map[string]time.Time), } diff --git a/internal/app/identity/app.go b/internal/app/identity/app.go index c22779a..de5f131 100644 --- a/internal/app/identity/app.go +++ b/internal/app/identity/app.go @@ -32,8 +32,10 @@ type Config struct { Sessions auth.SessionTokens // MaxScheduleSeconds 限制 self.update 的 default_delay_ms。 MaxScheduleSeconds int64 - // ConnControl 可选：logout 后踢线；未接线时为 nil。 + // ConnControl 可选：logout / 停用 / 删除后踢线；未接线时为 nil。 ConnControl port.ConnControl + // Downlink 可选：停用/删除时发 revoked 与群事件；未接线时为 nil。 + Downlink port.Downlink } // App 实现 identity.Service（含 I1 注册与 I2 self/对话密码）。 @@ -45,6 +47,7 @@ type App struct { sessions auth.SessionTokens maxScheduleSeconds int64 connCtrl port.ConnControl + down port.Downlink nowFn func() time.Time } @@ -81,6 +84,7 @@ func New(cfg Config) *App { sessions: cfg.Sessions, maxScheduleSeconds: cfg.MaxScheduleSeconds, connCtrl: cfg.ConnControl, + down: cfg.Downlink, nowFn: cfg.Now, } } diff --git a/internal/app/identity/lifecycle.go b/internal/app/identity/lifecycle.go new file mode 100644 index 0000000..745b0a3 --- /dev/null +++ b/internal/app/identity/lifecycle.go @@ -0,0 +1,664 @@ +package identity + +import ( + "bytes" + "context" + "database/sql" + "errors" + + "git.asio.asia/nixevol/NixMsg/internal/app/port" + "git.asio.asia/nixevol/NixMsg/internal/protocol" +) + +const ( + reasonEndpointDisabled = "endpoint_disabled" + reasonEndpointDeleted = "endpoint_deleted" + reasonSenderDisabled = "sender_disabled" + reasonSenderDeleted = "sender_deleted" + reasonLeftGroup = "left_group" + reasonGroupDissolved = "group_dissolved" + + eventOwnerChanged = "owner_changed" + eventLeft = "left" + eventMemberRemoved = "member_removed" + eventDissolved = "dissolved" +) + +type revokeItem struct { + endpointID string + msgID string + fromID string + reason string +} + +type groupNotify struct { + recipients []string + groupID string + event string + endpointID string + atMs int64 +} + +// Disable 停用端：清令牌、作废相关消息/投递，并踢连接（DEVELOPMENT 7.6）。 +func (a *App) Disable(ctx context.Context, endpointID string) error { + return a.disableOrDelete(ctx, endpointID, false) +} + +// Enable 仅恢复 enabled=1；已作废消息不恢复。 +func (a *App) Enable(ctx context.Context, endpointID string) error { + err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error { + res, e := tx.ExecContext(ctx, `UPDATE endpoints SET enabled = 1 WHERE id = ?`, endpointID) + if e != nil { + return e + } + n, _ := res.RowsAffected() + if n == 0 { + return errCode(protocol.CodeNotFound, "endpoint not found") + } + return nil + }) + return err +} + +// Delete 删除端：停用效果（原因改为 deleted）+ 退群/转让群主 + 清授权与发出记录。 +func (a *App) Delete(ctx context.Context, endpointID string) error { + return a.disableOrDelete(ctx, endpointID, true) +} + +func (a *App) disableOrDelete(ctx context.Context, endpointID string, hardDelete bool) error { + nowMs := a.now().UnixMilli() + recvReason := reasonEndpointDisabled + sendReason := reasonSenderDisabled + if hardDelete { + recvReason = reasonEndpointDeleted + sendReason = reasonSenderDeleted + } + + var revokes []revokeItem + var notifies []groupNotify + err := a.db.Queue.Do(ctx, func(tx *sql.Tx) error { + var one int + if e := tx.QueryRowContext(ctx, `SELECT 1 FROM endpoints WHERE id = ?`, endpointID).Scan(&one); e != nil { + if errors.Is(e, sql.ErrNoRows) { + return errCode(protocol.CodeNotFound, "endpoint not found") + } + return e + } + + if _, e := tx.ExecContext(ctx, ` +UPDATE endpoints SET enabled = 0, + session_hash = NULL, session_issued_at = NULL, session_used_at = NULL +WHERE id = ?`, endpointID); e != nil { + return e + } + + if e := voidEndpointMessagesTx(tx, endpointID, recvReason, sendReason, nowMs, &revokes); e != nil { + return e + } + + if !hardDelete { + return nil + } + + if e := leaveAllGroupsTx(tx, endpointID, nowMs, &revokes, ¬ifies); e != nil { + return e + } + if _, e := tx.ExecContext(ctx, `DELETE FROM talk_grants WHERE sender_id = ? OR target_id = ?`, endpointID, endpointID); e != nil { + return e + } + if _, e := tx.ExecContext(ctx, `DELETE FROM receipts WHERE sender_id = ?`, endpointID); e != nil { + return e + } + if _, e := tx.ExecContext(ctx, `DELETE FROM send_keys WHERE sender_id = ?`, endpointID); e != nil { + return e + } + if _, e := tx.ExecContext(ctx, `DELETE FROM messages WHERE sender_id = ?`, endpointID); e != nil { + return e + } + _, e := tx.ExecContext(ctx, `DELETE FROM endpoints WHERE id = ?`, endpointID) + return e + }) + if err != nil { + return err + } + + a.publishRevokes(ctx, revokes) + a.publishGroupEvents(ctx, notifies) + if a.connCtrl != nil { + _ = a.connCtrl.Disconnect(ctx, endpointID, "", port.DisconnectFatal) + } + return nil +} + +func voidEndpointMessagesTx(tx *sql.Tx, endpointID, recvReason, sendReason string, nowMs int64, revokes *[]revokeItem) error { + // 发给 X 的 pending → rejected + rows, err := tx.Query(` +SELECT d.seq, d.pushed_at, m.id, m.sender_id, m.receipt +FROM deliveries d +JOIN messages m ON m.seq = d.seq +WHERE d.endpoint_id = ? AND d.state = 'pending'`, endpointID) + if err != nil { + return err + } + type pendRow struct { + seq int64 + pushed sql.NullInt64 + msgID string + senderID string + receipt int + } + var pending []pendRow + for rows.Next() { + var r pendRow + if scanErr := rows.Scan(&r.seq, &r.pushed, &r.msgID, &r.senderID, &r.receipt); scanErr != nil { + _ = rows.Close() + return scanErr + } + pending = append(pending, r) + } + _ = rows.Close() + if err = rows.Err(); err != nil { + return err + } + finalSeqs := map[int64]struct{}{} + for _, r := range pending { + if _, execErr := tx.Exec(` +UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? +WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, + recvReason, nowMs, r.seq, endpointID); execErr != nil { + return execErr + } + if r.receipt != 0 { + if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, endpointID, "rejected", recvReason, nowMs, true); e != nil { + return e + } + } + if r.pushed.Valid && revokes != nil { + *revokes = append(*revokes, revokeItem{ + endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: recvReason, + }) + } + finalSeqs[r.seq] = struct{}{} + } + + // 发给 X 的 scheduled 单聊 → completed，要回执则写 + srows, err := tx.Query(` +SELECT seq, id, sender_id, receipt FROM messages +WHERE dest_kind = 'endpoint' AND dest_id = ? AND state = 'scheduled'`, endpointID) + if err != nil { + return err + } + type schedRow struct { + seq int64 + msgID string + senderID string + receipt int + } + var scheduledTo []schedRow + for srows.Next() { + var r schedRow + if scanErr := srows.Scan(&r.seq, &r.msgID, &r.senderID, &r.receipt); scanErr != nil { + _ = srows.Close() + return scanErr + } + scheduledTo = append(scheduledTo, r) + } + _ = srows.Close() + if err = srows.Err(); err != nil { + return err + } + for _, r := range scheduledTo { + if _, execErr := tx.Exec(` +UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`, + recvReason, r.seq); execErr != nil { + return execErr + } + if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, r.seq); execErr != nil { + return execErr + } + if r.receipt != 0 { + // 消息级作废：endpoint_id 空，state=rejected（DEVELOPMENT 6.4） + if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, "", "rejected", recvReason, nowMs, true); e != nil { + return e + } + } + } + + // X 发出的 scheduled → completed（sender_*），不写回执 + outSched, err := tx.Query(`SELECT seq FROM messages WHERE sender_id = ? AND state = 'scheduled'`, endpointID) + if err != nil { + return err + } + var outSeqs []int64 + for outSched.Next() { + var seq int64 + if scanErr := outSched.Scan(&seq); scanErr != nil { + _ = outSched.Close() + return scanErr + } + outSeqs = append(outSeqs, seq) + } + _ = outSched.Close() + if err = outSched.Err(); err != nil { + return err + } + for _, seq := range outSeqs { + if _, execErr := tx.Exec(` +UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`, + sendReason, seq); execErr != nil { + return execErr + } + if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq); execErr != nil { + return execErr + } + } + + // X 发出的消息的 pending 投递 → rejected（sender_*），不写回执 + drows, err := tx.Query(` +SELECT d.seq, d.endpoint_id, d.pushed_at, m.id, m.sender_id +FROM deliveries d +JOIN messages m ON m.seq = d.seq +WHERE m.sender_id = ? AND d.state = 'pending'`, endpointID) + if err != nil { + return err + } + type outPend struct { + seq int64 + endpointID string + pushed sql.NullInt64 + msgID string + senderID string + } + var outPending []outPend + for drows.Next() { + var r outPend + if scanErr := drows.Scan(&r.seq, &r.endpointID, &r.pushed, &r.msgID, &r.senderID); scanErr != nil { + _ = drows.Close() + return scanErr + } + outPending = append(outPending, r) + } + _ = drows.Close() + if err = drows.Err(); err != nil { + return err + } + for _, r := range outPending { + if _, execErr := tx.Exec(` +UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? +WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, + sendReason, nowMs, r.seq, r.endpointID); execErr != nil { + return execErr + } + if r.pushed.Valid && revokes != nil { + *revokes = append(*revokes, revokeItem{ + endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: sendReason, + }) + } + finalSeqs[r.seq] = struct{}{} + } + + for seq := range finalSeqs { + if e := tryFinalizeTx(tx, seq); e != nil { + return e + } + } + return nil +} + +func leaveAllGroupsTx(tx *sql.Tx, endpointID string, nowMs int64, revokes *[]revokeItem, notifies *[]groupNotify) error { + grows, err := tx.Query(` +SELECT g.id, g.owner_id +FROM groups g +JOIN group_members gm ON gm.group_id = g.id +WHERE gm.endpoint_id = ?`, endpointID) + if err != nil { + return err + } + type grow struct { + id string + owner string + } + var groups []grow + for grows.Next() { + var g grow + if scanErr := grows.Scan(&g.id, &g.owner); scanErr != nil { + _ = grows.Close() + return scanErr + } + groups = append(groups, g) + } + _ = grows.Close() + if err = grows.Err(); err != nil { + return err + } + + for _, g := range groups { + members, memErr := loadMembersTx(tx, g.id) + if memErr != nil { + return memErr + } + if g.owner == endpointID { + others := withoutMember(members, endpointID) + if len(others) == 0 { + if e := voidGroupAllTx(tx, g.id, nowMs, revokes); e != nil { + return e + } + if _, e := tx.Exec(`DELETE FROM group_members WHERE group_id = ?`, g.id); e != nil { + return e + } + if _, e := tx.Exec(`DELETE FROM groups WHERE id = ?`, g.id); e != nil { + return e + } + if notifies != nil { + *notifies = append(*notifies, groupNotify{ + recipients: members, groupID: g.id, event: eventDissolved, atMs: nowMs, + }) + } + continue + } + newOwner, ownErr := earliestOtherMemberTx(tx, g.id, endpointID) + if ownErr != nil { + return ownErr + } + if _, e := tx.Exec(`UPDATE groups SET owner_id = ? WHERE id = ?`, newOwner, g.id); e != nil { + return e + } + if _, e := tx.Exec(`DELETE FROM group_members WHERE group_id = ? AND endpoint_id = ?`, g.id, endpointID); e != nil { + return e + } + if e := voidMemberDeliveriesTx(tx, g.id, endpointID, reasonLeftGroup, nowMs, revokes); e != nil { + return e + } + left := withoutMember(members, endpointID) + if notifies != nil { + *notifies = append(*notifies, + groupNotify{recipients: left, groupID: g.id, event: eventOwnerChanged, endpointID: newOwner, atMs: nowMs}, + groupNotify{recipients: append(append([]string{}, left...), endpointID), groupID: g.id, event: eventMemberRemoved, endpointID: endpointID, atMs: nowMs}, + ) + } + continue + } + + if _, e := tx.Exec(`DELETE FROM group_members WHERE group_id = ? AND endpoint_id = ?`, g.id, endpointID); e != nil { + return e + } + if e := voidMemberDeliveriesTx(tx, g.id, endpointID, reasonLeftGroup, nowMs, revokes); e != nil { + return e + } + left := withoutMember(members, endpointID) + if notifies != nil { + *notifies = append(*notifies, groupNotify{ + recipients: append(append([]string{}, left...), endpointID), + groupID: g.id, event: eventLeft, endpointID: endpointID, atMs: nowMs, + }) + } + } + return nil +} + +func loadMembersTx(tx *sql.Tx, groupID string) ([]string, error) { + rows, err := tx.Query(`SELECT endpoint_id FROM group_members WHERE group_id = ? ORDER BY joined_at ASC, endpoint_id ASC`, groupID) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + var out []string + for rows.Next() { + var id string + if e := rows.Scan(&id); e != nil { + return nil, e + } + out = append(out, id) + } + return out, rows.Err() +} + +func earliestOtherMemberTx(tx *sql.Tx, groupID, exceptID string) (string, error) { + var id string + err := tx.QueryRow(` +SELECT endpoint_id FROM group_members +WHERE group_id = ? AND endpoint_id != ? +ORDER BY joined_at ASC, endpoint_id ASC +LIMIT 1`, groupID, exceptID).Scan(&id) + return id, err +} + +func withoutMember(ids []string, drop string) []string { + out := make([]string, 0, len(ids)) + for _, id := range ids { + if id != drop { + out = append(out, id) + } + } + return out +} + +// voidMemberDeliveriesTx 与 group 包同语义：退群成员的 pending 群投递改 rejected。 +func voidMemberDeliveriesTx(tx *sql.Tx, groupID, endpointID, reason string, nowMs int64, revokes *[]revokeItem) error { + rows, err := tx.Query(` +SELECT d.seq, d.pushed_at, m.id, m.sender_id +FROM deliveries d +JOIN messages m ON m.seq = d.seq +WHERE d.endpoint_id = ? AND d.state = 'pending' + AND m.dest_kind = 'group' AND m.dest_id = ?`, endpointID, groupID) + if err != nil { + return err + } + type row struct { + seq int64 + pushed sql.NullInt64 + msgID string + senderID string + } + var list []row + for rows.Next() { + var r row + if scanErr := rows.Scan(&r.seq, &r.pushed, &r.msgID, &r.senderID); scanErr != nil { + _ = rows.Close() + return scanErr + } + list = append(list, r) + } + _ = rows.Close() + if err = rows.Err(); err != nil { + return err + } + for _, r := range list { + if _, execErr := tx.Exec(` +UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? +WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, + reason, nowMs, r.seq, endpointID); execErr != nil { + return execErr + } + if r.pushed.Valid && revokes != nil { + *revokes = append(*revokes, revokeItem{ + endpointID: endpointID, msgID: r.msgID, fromID: r.senderID, reason: reason, + }) + } + if e := tryFinalizeTx(tx, r.seq); e != nil { + return e + } + } + return nil +} + +func voidGroupAllTx(tx *sql.Tx, groupID string, nowMs int64, revokes *[]revokeItem) error { + rows, err := tx.Query(` +SELECT d.seq, d.endpoint_id, d.pushed_at, m.id, m.sender_id +FROM deliveries d +JOIN messages m ON m.seq = d.seq +WHERE d.state = 'pending' AND m.dest_kind = 'group' AND m.dest_id = ?`, groupID) + if err != nil { + return err + } + type drow struct { + seq int64 + endpointID string + pushed sql.NullInt64 + msgID string + senderID string + } + var dlist []drow + for rows.Next() { + var r drow + if scanErr := rows.Scan(&r.seq, &r.endpointID, &r.pushed, &r.msgID, &r.senderID); scanErr != nil { + _ = rows.Close() + return scanErr + } + dlist = append(dlist, r) + } + _ = rows.Close() + if err = rows.Err(); err != nil { + return err + } + for _, r := range dlist { + if _, execErr := tx.Exec(` +UPDATE deliveries SET state = 'rejected', reason = ?, updated_at = ? +WHERE seq = ? AND endpoint_id = ? AND state = 'pending'`, + reasonGroupDissolved, nowMs, r.seq, r.endpointID); execErr != nil { + return execErr + } + if r.pushed.Valid && revokes != nil { + *revokes = append(*revokes, revokeItem{ + endpointID: r.endpointID, msgID: r.msgID, fromID: r.senderID, reason: reasonGroupDissolved, + }) + } + if e := tryFinalizeTx(tx, r.seq); e != nil { + return e + } + } + + srows, err := tx.Query(` +SELECT seq, id, sender_id, receipt FROM messages +WHERE dest_kind = 'group' AND dest_id = ? AND state = 'scheduled'`, groupID) + if err != nil { + return err + } + type srow struct { + seq int64 + msgID string + senderID string + receipt int + } + var slist []srow + for srows.Next() { + var r srow + if scanErr := srows.Scan(&r.seq, &r.msgID, &r.senderID, &r.receipt); scanErr != nil { + _ = srows.Close() + return scanErr + } + slist = append(slist, r) + } + _ = srows.Close() + if err = srows.Err(); err != nil { + return err + } + for _, r := range slist { + if _, execErr := tx.Exec(` +UPDATE messages SET state = 'completed', reason = ? WHERE seq = ? AND state = 'scheduled'`, + reasonGroupDissolved, r.seq); execErr != nil { + return execErr + } + if _, execErr := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, r.seq); execErr != nil { + return execErr + } + if r.receipt != 0 { + if e := insertReceiptIfWantedTx(tx, r.senderID, r.msgID, "", "rejected", reasonGroupDissolved, nowMs, true); e != nil { + return e + } + } + } + return nil +} + +func insertReceiptIfWantedTx(tx *sql.Tx, senderID, msgID, endpointID, state, reason string, nowMs int64, alreadyWanted bool) error { + if !alreadyWanted { + return nil + } + var one int + err := tx.QueryRow(`SELECT 1 FROM endpoints WHERE id = ?`, senderID).Scan(&one) + if errors.Is(err, sql.ErrNoRows) { + return nil + } + if err != nil { + return err + } + _, err = tx.Exec(` +INSERT INTO receipts(sender_id, msg_id, endpoint_id, state, reason, created_at, acked) +VALUES(?,?,?,?,?,?,0)`, senderID, msgID, endpointID, state, reason, nowMs) + return err +} + +func tryFinalizeTx(tx *sql.Tx, seq int64) error { + var n int + if err := tx.QueryRow(`SELECT COUNT(*) FROM deliveries WHERE seq = ? AND state = 'pending'`, seq).Scan(&n); err != nil { + return err + } + if n > 0 { + return nil + } + var state string + if err := tx.QueryRow(`SELECT state FROM messages WHERE seq = ?`, seq).Scan(&state); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil + } + return err + } + if state == "completed" { + return nil + } + if _, err := tx.Exec(`UPDATE messages SET state = 'completed' WHERE seq = ?`, seq); err != nil { + return err + } + _, err := tx.Exec(`DELETE FROM message_bodies WHERE seq = ?`, seq) + return err +} + +func (a *App) publishRevokes(ctx context.Context, items []revokeItem) { + if a.down == nil || len(items) == 0 { + return + } + for _, it := range items { + frame := protocol.Revoked{ + V: protocol.Version, Type: protocol.TypeRevoked, + ID: it.msgID, From: it.fromID, Reason: it.reason, + } + payload, encErr := encodeFrame(frame) + if encErr != nil { + continue + } + _ = a.down.PublishDown(ctx, it.endpointID, "", payload, port.PublishOpts{QoS: 1}) + } +} + +func (a *App) publishGroupEvents(ctx context.Context, items []groupNotify) { + if a.down == nil || len(items) == 0 { + return + } + for _, it := range items { + frame := protocol.GroupEvent{ + V: protocol.Version, Type: protocol.TypeGroupEvent, + GroupID: it.groupID, Event: it.event, EndpointID: it.endpointID, AtMs: it.atMs, + } + payload, encErr := encodeFrame(frame) + if encErr != nil { + continue + } + seen := map[string]struct{}{} + for _, id := range it.recipients { + if _, ok := seen[id]; ok { + continue + } + seen[id] = struct{}{} + _ = a.down.PublishDown(ctx, id, "", payload, port.PublishOpts{QoS: 0}) + } + } +} + +func encodeFrame(v any) ([]byte, error) { + var buf bytes.Buffer + if err := protocol.Encode(&buf, v); err != nil { + return nil, err + } + return buf.Bytes(), nil +} diff --git a/internal/app/identity/lifecycle_test.go b/internal/app/identity/lifecycle_test.go new file mode 100644 index 0000000..882ef09 --- /dev/null +++ b/internal/app/identity/lifecycle_test.go @@ -0,0 +1,386 @@ +package identity_test + +import ( + "context" + "database/sql" + "io" + "net/http" + "net/http/cookiejar" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" + + "git.asio.asia/nixevol/NixMsg/internal/admin" + "git.asio.asia/nixevol/NixMsg/internal/app/identity" + "git.asio.asia/nixevol/NixMsg/internal/app/message" + "git.asio.asia/nixevol/NixMsg/internal/app/port" + "git.asio.asia/nixevol/NixMsg/internal/auth" + "git.asio.asia/nixevol/NixMsg/internal/config" + "git.asio.asia/nixevol/NixMsg/internal/protocol" + "git.asio.asia/nixevol/NixMsg/internal/store" +) + +func openLifecycle(t *testing.T) (*identity.App, *message.App, *store.DB) { + t.Helper() + dir := t.TempDir() + db, err := store.Open(filepath.Join(dir, "data"), "FULL") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + fixed := time.UnixMilli(1_700_000_000_000) + idApp := identity.New(identity.Config{ + DB: db, + Hash: auth.NewStubHashPool(), + Locks: auth.NewStubLoginLocks(), + Sessions: auth.NewSessionTokens(), + MaxScheduleSeconds: int64(config.Default().Limits.MaxScheduleSeconds), + Now: func() time.Time { return fixed }, + ConnControl: &port.StubConnControl{}, + Downlink: &port.StubDownlink{}, + }) + lim := message.LimitsFromConfig(config.Default().Limits) + lim.RequestsPerSecond = 0 + lim.RecordRetentionDays = 7 + msgApp := message.New(db, lim, auth.NewStubHashPool(), + message.WithNow(func() time.Time { return fixed }), + message.WithLocks(auth.NewStubLoginLocks()), + ) + return idApp, msgApp, db +} + +func insertEPFull(t *testing.T, db *store.DB, id string) { + t.Helper() + err := db.Queue.Do(context.Background(), func(tx *sql.Tx) error { + _, e := tx.Exec(` +INSERT INTO endpoints(id, name, login_hash, talk_hash, talk_version, default_delay_ms, enabled, created_at, offline_since) +VALUES(?,?,?,?,0,0,1,?,?)`, id, id, "stub$login", nil, 1_700_000_000_000, 1_700_000_000_000) + return e + }) + if err != nil { + t.Fatal(err) + } +} + +func TestF01DisableVoidsScheduledAndRejectsNew(t *testing.T) { + t.Parallel() + idApp, msgApp, db := openLifecycle(t) + ctx := context.Background() + insertEPFull(t, db, "alice") + insertEPFull(t, db, "bob") + + future := int64(1_700_000_000_000 + 3600_000) + receipt := true + toBob := &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "m-to-bob", + To: protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "hi"}, + SendAtMs: &future, + Receipt: &receipt, + } + if _, err := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, toBob); err != nil { + t.Fatal(err) + } + fromBob := &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "2", ID: "m-from-bob", + To: protocol.Target{Kind: protocol.TargetEndpoint, ID: "alice"}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "bye"}, + SendAtMs: &future, + } + if _, err := msgApp.Submit(ctx, "bob", port.ConnInfo{EndpointID: "bob"}, fromBob); err != nil { + t.Fatal(err) + } + + if err := idApp.Disable(ctx, "bob"); err != nil { + t.Fatal(err) + } + + var enabled int + if err := db.Read.QueryRow(`SELECT enabled FROM endpoints WHERE id='bob'`).Scan(&enabled); err != nil { + t.Fatal(err) + } + if enabled != 0 { + t.Fatalf("enabled=%d", enabled) + } + + var toState, toReason string + if err := db.Read.QueryRow(`SELECT state, reason FROM messages WHERE sender_id='alice' AND id='m-to-bob'`). + Scan(&toState, &toReason); err != nil { + t.Fatal(err) + } + if toState != "completed" || toReason != "endpoint_disabled" { + t.Fatalf("to bob: state=%s reason=%s", toState, toReason) + } + var fromState, fromReason string + if err := db.Read.QueryRow(`SELECT state, reason FROM messages WHERE sender_id='bob' AND id='m-from-bob'`). + Scan(&fromState, &fromReason); err != nil { + t.Fatal(err) + } + if fromState != "completed" || fromReason != "sender_disabled" { + t.Fatalf("from bob: state=%s reason=%s", fromState, fromReason) + } + + newSend := &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "3", ID: "m-new", + To: protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "x"}, + } + _, err := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, newSend) + if protoCode(err) != protocol.CodeEndpointDisabled { + t.Fatalf("want endpoint_disabled got %v", err) + } + + if err := idApp.Enable(ctx, "bob"); err != nil { + t.Fatal(err) + } + if err := db.Read.QueryRow(`SELECT state FROM messages WHERE id='m-to-bob'`).Scan(&toState); err != nil { + t.Fatal(err) + } + if toState != "completed" { + t.Fatalf("voided message restored? %s", toState) + } + okSend := &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "4", ID: "m-after", + To: protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "ok"}, + } + if _, err := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, okSend); err != nil { + t.Fatal(err) + } +} + +func TestF01DeleteOwnerTransfersEarliest(t *testing.T) { + t.Parallel() + idApp, _, db := openLifecycle(t) + ctx := context.Background() + insertEPFull(t, db, "owner") + insertEPFull(t, db, "early") + insertEPFull(t, db, "late") + + err := db.Queue.Do(ctx, func(tx *sql.Tx) error { + if _, e := tx.Exec(`INSERT INTO groups(id, name, owner_id, created_at) VALUES('g1','群','owner',?)`, 1_700_000_000_000); e != nil { + return e + } + _, e := tx.Exec(`INSERT INTO group_members(group_id, endpoint_id, joined_at) VALUES +('g1','owner',100),('g1','early',200),('g1','late',300)`) + return e + }) + if err != nil { + t.Fatal(err) + } + + if err := idApp.Delete(ctx, "owner"); err != nil { + t.Fatal(err) + } + + var owner string + if err := db.Read.QueryRow(`SELECT owner_id FROM groups WHERE id='g1'`).Scan(&owner); err != nil { + t.Fatal(err) + } + if owner != "early" { + t.Fatalf("want earliest other member early, got %s", owner) + } + var n int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM group_members WHERE group_id='g1' AND endpoint_id='owner'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatal("owner should have left the group") + } +} + +func TestF01DeleteReopenNoOldReceipts(t *testing.T) { + t.Parallel() + idApp, msgApp, db := openLifecycle(t) + ctx := context.Background() + insertEPFull(t, db, "alice") + insertEPFull(t, db, "bob") + + receipt := true + send := &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "m1", + To: protocol.Target{Kind: protocol.TargetEndpoint, ID: "alice"}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "hi"}, + Receipt: &receipt, + } + if _, err := msgApp.Submit(ctx, "bob", port.ConnInfo{EndpointID: "bob"}, send); err != nil { + t.Fatal(err) + } + err := db.Queue.Do(ctx, func(tx *sql.Tx) error { + _, e := tx.Exec(` +INSERT INTO receipts(sender_id, msg_id, endpoint_id, state, reason, created_at, acked) +VALUES('bob','m1','alice','accepted','',?,0)`, 1_700_000_000_000) + return e + }) + if err != nil { + t.Fatal(err) + } + + if err := idApp.Delete(ctx, "bob"); err != nil { + t.Fatal(err) + } + + var n int + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE sender_id='bob'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("old receipts should be gone, got %d", n) + } + + insertEPFull(t, db, "bob") + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM receipts WHERE sender_id='bob'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("reopened endpoint must not inherit receipts, got %d", n) + } + if err := db.Read.QueryRow(`SELECT COUNT(*) FROM messages WHERE sender_id='bob'`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("reopened endpoint must not inherit messages, got %d", n) + } +} + +func TestAdminDisableDeleteHTTP(t *testing.T) { + t.Parallel() + dir := t.TempDir() + db, err := store.Open(filepath.Join(dir, "data"), "FULL") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + + fixed := time.UnixMilli(1_700_000_000_000) + hash := auth.NewStubHashPool() + if seedErr := admin.SeedAdminPassword(context.Background(), db, hash, "adminpassword1"); seedErr != nil { + t.Fatal(seedErr) + } + idApp := identity.New(identity.Config{ + DB: db, Hash: hash, Locks: auth.NewStubLoginLocks(), Sessions: auth.NewSessionTokens(), + MaxScheduleSeconds: 86400, Now: func() time.Time { return fixed }, + }) + lim := message.LimitsFromConfig(config.Default().Limits) + lim.RequestsPerSecond = 0 + msgApp := message.New(db, lim, hash, + message.WithNow(func() time.Time { return fixed }), + message.WithLocks(auth.NewStubLoginLocks()), + ) + + kick := &lifecycleKick{} + h := admin.New(admin.Deps{ + DB: db, Hash: hash, Tokens: admin.NewRandomAPITokens(), + Locks: admin.NewMemoryLoginLocks(), KickEndpoint: kick.Kick, Identity: idApp, + }) + srv := httptest.NewServer(h) + t.Cleanup(srv.Close) + + jar, _ := cookiejar.New(nil) + client := &http.Client{Jar: jar} + loginRes, err := client.Post(srv.URL+"/api/admin/login", "application/json", + strings.NewReader(`{"username":"admin","password":"adminpassword1"}`)) + if err != nil { + t.Fatal(err) + } + _ = loginRes.Body.Close() + if loginRes.StatusCode != 200 { + t.Fatalf("login %d", loginRes.StatusCode) + } + + createEP := func(id string) { + t.Helper() + req, _ := http.NewRequest(http.MethodPost, srv.URL+"/api/admin/endpoints", + strings.NewReader(`{"id":"`+id+`","name":"`+id+`","login_password":"password12"}`)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Nixmsg-Request", "1") + res, e := client.Do(req) + if e != nil { + t.Fatal(e) + } + raw, _ := io.ReadAll(res.Body) + _ = res.Body.Close() + if res.StatusCode != 200 { + t.Fatalf("create %s: %d %s", id, res.StatusCode, raw) + } + } + createEP("alice") + createEP("bob") + createEP("carol") + + ctx := context.Background() + future := fixed.UnixMilli() + 3600_000 + toBob := &protocol.Send{ + V: protocol.Version, Type: protocol.TypeSend, RID: "1", ID: "sched-bob", + To: protocol.Target{Kind: protocol.TargetEndpoint, ID: "bob"}, + Body: protocol.Body{Enc: protocol.EncUTF8, Data: "x"}, + SendAtMs: &future, + } + if _, subErr := msgApp.Submit(ctx, "alice", port.ConnInfo{EndpointID: "alice"}, toBob); subErr != nil { + t.Fatal(subErr) + } + + req, _ := http.NewRequest(http.MethodPost, srv.URL+"/api/admin/endpoints/batch", + strings.NewReader(`{"ids":["bob"],"action":"disable"}`)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Nixmsg-Request", "1") + res, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(res.Body) + _ = res.Body.Close() + if res.StatusCode != 200 { + t.Fatalf("disable: %d %s", res.StatusCode, body) + } + var st string + if scanErr := db.Read.QueryRow(`SELECT state FROM messages WHERE id='sched-bob'`).Scan(&st); scanErr != nil { + t.Fatal(scanErr) + } + if st != "completed" { + t.Fatalf("scheduled should be voided, got %s", st) + } + + err = db.Queue.Do(ctx, func(tx *sql.Tx) error { + if _, e := tx.Exec(`INSERT INTO groups(id, name, owner_id, created_at) VALUES('ghttp','G','bob',?)`, fixed.UnixMilli()); e != nil { + return e + } + _, e := tx.Exec(`INSERT INTO group_members(group_id, endpoint_id, joined_at) VALUES +('ghttp','bob',1),('ghttp','alice',2),('ghttp','carol',3)`) + return e + }) + if err != nil { + t.Fatal(err) + } + + // bob 已停用，需先启用才能作为「仍存在的群主」再删除？删除不要求 enabled。 + req, _ = http.NewRequest(http.MethodDelete, srv.URL+"/api/admin/endpoints/bob", nil) + req.Header.Set("X-Nixmsg-Request", "1") + res, err = client.Do(req) + if err != nil { + t.Fatal(err) + } + raw, _ := io.ReadAll(res.Body) + _ = res.Body.Close() + if res.StatusCode != 200 { + t.Fatalf("delete: %d %s", res.StatusCode, raw) + } + var owner string + if err := db.Read.QueryRow(`SELECT owner_id FROM groups WHERE id='ghttp'`).Scan(&owner); err != nil { + t.Fatal(err) + } + if owner != "alice" { + t.Fatalf("want alice as new owner, got %s", owner) + } +} + +type lifecycleKick struct { + Calls []string +} + +func (k *lifecycleKick) Kick(_ context.Context, endpointID string) (bool, error) { + k.Calls = append(k.Calls, endpointID) + return true, nil +} diff --git a/internal/app/identity/self.go b/internal/app/identity/self.go index 912039d..fa75a76 100644 --- a/internal/app/identity/self.go +++ b/internal/app/identity/self.go @@ -215,8 +215,3 @@ WHERE id = ?`, endpointID) } return nil } - -// Disable / Enable / Delete 属 I5，此处保留未实现。 -func (a *App) Disable(context.Context, string) error { return ErrNotImplemented } -func (a *App) Enable(context.Context, string) error { return ErrNotImplemented } -func (a *App) Delete(context.Context, string) error { return ErrNotImplemented }