fix: 停用删除重置密码发 fatal 并注入 Downlink 发 revoked

internal/app/identity/lifecycle_test.go

@@ -3,6 +3,7 @@ package identity_test
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"io"
 	"net/http"
 	"net/http/cookiejar"
@@ -64,6 +65,82 @@ VALUES(?,?,?,?,0,0,1,?,?)`, id, id, "stub$login", nil, 1_700_000_000_000, 1_700_
 	}
 }
 
+func TestDisableEmitsRevokedForPushed(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)
+	down := &message.RecordingDownlink{}
+	ctrl := &port.StubConnControl{}
+	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:        ctrl,
+		Downlink:           down,
+	})
+	ctx := context.Background()
+	insertEPFull(t, db, "alice")
+	insertEPFull(t, db, "bob")
+
+	err = db.Queue.Do(ctx, func(tx *sql.Tx) error {
+		res, e := tx.Exec(`
+INSERT INTO messages(
+  id, sender_id, dest_kind, dest_id, meta, content_type, body_enc,
+  send_at, keep, ttl_seconds, receipt, state, reason, created_at)
+VALUES('pushed-1','alice','endpoint','bob','{}','text/plain','utf8',?,1,0,0,'dispatched','',?)`,
+			fixed.UnixMilli(), fixed.UnixMilli())
+		if e != nil {
+			return e
+		}
+		seq, _ := res.LastInsertId()
+		_, e = tx.Exec(`
+INSERT INTO deliveries(seq, endpoint_id, send_at, keep, state, reason, updated_at, pushed_at, pushed_conn)
+VALUES(?,?,?,1,'pending','',?,?,?)`,
+			seq, "bob", fixed.UnixMilli(), fixed.UnixMilli(), fixed.UnixMilli(), "c-bob")
+		return e
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := idApp.Disable(ctx, "bob"); err != nil {
+		t.Fatal(err)
+	}
+
+	if down.FilterType(protocol.TypeRevoked) != 1 {
+		t.Fatalf("want 1 revoked, got snapshots=%v", down.Snapshots())
+	}
+	p := down.Snapshots()[0]
+	var head struct {
+		Type   string `json:"type"`
+		Reason string `json:"reason"`
+		ID     string `json:"id"`
+	}
+	_ = json.Unmarshal(p.Payload, &head)
+	if head.Type != protocol.TypeRevoked || head.Reason != "endpoint_disabled" || head.ID != "pushed-1" {
+		t.Fatalf("revoked=%+v", head)
+	}
+	if p.EndpointID != "bob" || p.QoS != 1 {
+		t.Fatalf("publish=%+v", p)
+	}
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if len(ctrl.Calls) == 1 && ctrl.Calls[0] == "bob" {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("disconnect calls=%v", ctrl.Calls)
+}
+
 func TestF01DisableVoidsScheduledAndRejectsNew(t *testing.T) {
 	t.Parallel()
 	idApp, msgApp, db := openLifecycle(t)