package presence_test import ( "context" "database/sql" "path/filepath" "sync" "testing" "time" "git.asio.asia/nixevol/NixMsg/internal/app/port" "git.asio.asia/nixevol/NixMsg/internal/app/presence" "git.asio.asia/nixevol/NixMsg/internal/protocol" "git.asio.asia/nixevol/NixMsg/internal/store" ) type memDownlink struct { mu sync.Mutex msgs []downMsg } type downMsg struct { endpointID string payload []byte qos byte } func (d *memDownlink) PublishDown(_ context.Context, endpointID string, _ port.ConnID, payload []byte, opts port.PublishOpts) error { d.mu.Lock() defer d.mu.Unlock() cp := append([]byte(nil), payload...) d.msgs = append(d.msgs, downMsg{endpointID: endpointID, payload: cp, qos: opts.QoS}) return nil } func (d *memDownlink) take() []downMsg { d.mu.Lock() defer d.mu.Unlock() out := d.msgs d.msgs = nil return out } func openPresence(t *testing.T) (*presence.App, *store.DB, *memDownlink) { t.Helper() db, err := store.Open(filepath.Join(t.TempDir(), "data"), "FULL") if err != nil { t.Fatal(err) } t.Cleanup(func() { _ = db.Close() }) down := &memDownlink{} fixed := time.UnixMilli(1_700_000_000_000) app := presence.New(presence.Config{ DB: db, Downlink: down, Now: func() time.Time { return fixed }, }) return app, db, down } func insertEP(t *testing.T, db *store.DB, id, name 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) VALUES(?,?,?,?,0,0,1,?)`, id, name, "stub$login", nil, 1_700_000_000_000) return e }) if err != nil { t.Fatal(err) } } func TestF03PresenceAndDirectory(t *testing.T) { t.Parallel() app, db, _ := openPresence(t) ctx := context.Background() insertEP(t, db, "alice", "Alice") insertEP(t, db, "bob", "Bob") insertEP(t, db, "carol", "Carol") items, err := app.Get(ctx, []string{"alice", "nobody"}) if err != nil { t.Fatal(err) } if len(items) != 2 || items[0].Online || !items[1].NotFound { t.Fatalf("%+v", items) } if err = app.SetOnline(ctx, "alice", "c1", 1_700_000_000_100); err != nil { t.Fatal(err) } items, _ = app.Get(ctx, []string{"alice"}) if !items[0].Online || items[0].SinceMs != 1_700_000_000_100 { t.Fatalf("%+v", items[0]) } // 正常断开后查为离线 if err = app.SetOffline(ctx, "alice", "c1", 1_700_000_000_200); err != nil { t.Fatal(err) } items, _ = app.Get(ctx, []string{"alice"}) if items[0].Online { t.Fatal("should be offline") } dir, next, err := app.Directory(ctx, &protocol.DirectoryList{ V: protocol.Version, Type: protocol.TypeDirectoryList, RID: "1", Limit: 2, }) if err != nil || len(dir) != 2 || next == "" { t.Fatalf("dir=%+v next=%q err=%v", dir, next, err) } dir2, next2, err := app.Directory(ctx, &protocol.DirectoryList{ V: protocol.Version, Type: protocol.TypeDirectoryList, RID: "2", Cursor: next, Limit: 10, }) if err != nil || len(dir2) != 1 || next2 != "" { t.Fatalf("dir2=%+v next=%q", dir2, next2) } // query:编号前缀 q, _, err := app.Directory(ctx, &protocol.DirectoryList{ V: protocol.Version, Type: protocol.TypeDirectoryList, RID: "3", Query: "bo", Limit: 10, }) if err != nil || len(q) != 1 || q[0].ID != "bob" { t.Fatalf("%+v err=%v", q, err) } // 名称包含不区分大小写 q, _, err = app.Directory(ctx, &protocol.DirectoryList{ V: protocol.Version, Type: protocol.TypeDirectoryList, RID: "4", Query: "car", Limit: 10, }) if err != nil || len(q) != 1 || q[0].ID != "carol" { t.Fatalf("%+v", q) } } func TestF04PresenceWatch(t *testing.T) { t.Parallel() app, db, down := openPresence(t) ctx := context.Background() insertEP(t, db, "alice", "A") insertEP(t, db, "bob", "B") insertEP(t, db, "carol", "C") if err := app.Watch(ctx, "conn-sub", "watcher", &protocol.PresenceWatch{ V: protocol.Version, Type: protocol.TypePresenceWatch, RID: "1", IDs: []string{"alice"}, }); err != nil { t.Fatal(err) } _ = app.SetOnline(ctx, "alice", "ca", 100) _ = app.SetOffline(ctx, "alice", "ca", 200) _ = app.SetOnline(ctx, "bob", "cb", 300) // 未订阅 msgs := down.take() if len(msgs) != 2 { t.Fatalf("want 2 presence for alice got %d", len(msgs)) } for _, m := range msgs { if m.endpointID != "watcher" || m.qos != 0 { t.Fatalf("%+v", m) } } // 再次 Watch 覆盖;断线清空 if err := app.Watch(ctx, "conn-sub", "watcher", &protocol.PresenceWatch{ V: protocol.Version, Type: protocol.TypePresenceWatch, RID: "2", All: true, }); err != nil { t.Fatal(err) } down.take() _ = app.SetOnline(ctx, "carol", "cc", 400) if n := len(down.take()); n != 1 { t.Fatalf("all watch got %d", n) } app.ClearWatch("conn-sub") _ = app.SetOffline(ctx, "carol", "cc", 500) if n := len(down.take()); n != 0 { t.Fatalf("after clear got %d", n) } }