package store import ( "context" "errors" "io" "net" "strings" "sync" "testing" "time" "github.com/alicebob/miniredis/v2" "github.com/redis/go-redis/v9" "migadu_counters/internal/config" ) func newMini(t *testing.T) *miniredis.Miniredis { t.Helper() m := miniredis.RunT(t) return m } // testStore wires a Store to two separate miniredis instances so tests can // prove which side (write master vs read replica) a method talks to. func testStore(t *testing.T) (*Store, *miniredis.Miniredis, *miniredis.Miniredis) { t.Helper() w, r := newMini(t), newMini(t) s := NewWithClients( redis.NewClient(&redis.Options{Addr: w.Addr()}), []*redis.Client{redis.NewClient(&redis.Options{Addr: r.Addr()})}, ) t.Cleanup(func() { s.Close() }) return s, w, r } func TestKey(t *testing.T) { tests := []struct { level, id, date, flow, want string }{ {"A", "x@example.com", "2026-10-13", "i", "cnt/A/x@example.com/d/2026-10-13/i"}, {"D", "example.com", "2026-10-13", "o", "cnt/D/example.com/d/2026-10-13/o"}, {"O", "1234", "2026-01-01", "i", "cnt/O/1234/d/2026-01-01/i"}, } for _, tt := range tests { if got := Key(tt.level, tt.id, tt.date, tt.flow); got != tt.want { t.Errorf("Key(%s,%s,%s,%s) = %q, want %q", tt.level, tt.id, tt.date, tt.flow, got, tt.want) } } } // The day stamp is always the UTC calendar day, matching migadu_web's // Date.utc_today() — never the server's local day. func TestDayKey_UTC(t *testing.T) { cest := time.FixedZone("CEST", 2*3600) tests := []struct { in time.Time want string }{ {time.Date(2026, 10, 13, 23, 59, 0, 0, cest), "2026-10-13"}, // 21:59 UTC {time.Date(2026, 10, 14, 1, 30, 0, 0, cest), "2026-10-13"}, // 23:30 UTC the day before {time.Date(2026, 10, 14, 2, 0, 0, 0, cest), "2026-10-14"}, // 00:00 UTC {time.Date(2026, 3, 5, 12, 0, 0, 0, time.UTC), "2026-03-05"}, // zero padded } for _, tt := range tests { if got := DayKey(tt.in); got != tt.want { t.Errorf("DayKey(%v) = %q, want %q", tt.in, got, tt.want) } } } func TestIncr(t *testing.T) { s, w, _ := testStore(t) ctx := context.Background() keys := []string{"cnt/A/x@example.com/d/2026-10-13/o", "cnt/O/1234/d/2026-10-13/o"} got, err := s.Incr(ctx, keys) if err != nil { t.Fatalf("Incr: %v", err) } if len(got) != 2 || got[0].Count != 1 || got[1].Count != 1 || got[0].Err != nil || got[1].Err != nil { t.Fatalf("first Incr = %+v, want counts [1 1]", got) } got, err = s.Incr(ctx, keys) if err != nil || got[0].Count != 2 || got[1].Count != 2 { t.Fatalf("second Incr = %+v, %v; want [2 2]", got, err) } for _, k := range keys { if v, _ := w.Get(k); v != "2" { t.Errorf("stored %s = %q, want \"2\"", k, v) } if ttl := w.TTL(k); ttl <= 30*24*time.Hour || ttl > TTL { t.Errorf("TTL(%s) = %v, want (30d, 31d]", k, ttl) } } if got, err := s.Incr(ctx, nil); err != nil || len(got) != 0 { t.Errorf("Incr(no keys) = %v, %v; want empty, nil", got, err) } } // One key holding a non-number fails its own INCR inside EXEC; the other // keys are still incremented and their counts reported (review #4). func TestIncr_PartialFailure(t *testing.T) { s, w, _ := testStore(t) w.Set("bad", "garbage") got, err := s.Incr(context.Background(), []string{"k1", "bad", "k3"}) if err != nil { t.Fatalf("a per-key failure must not fail the whole Incr: %v", err) } if got[0].Count != 1 || got[0].Err != nil || got[2].Count != 1 || got[2].Err != nil { t.Errorf("good keys = %+v %+v, want count 1 without error", got[0], got[2]) } if got[1].Err == nil || !strings.Contains(got[1].Err.Error(), "not an integer") { t.Errorf("bad key err = %v, want the INCR error", got[1].Err) } if v, _ := w.Get("bad"); v != "garbage" { t.Errorf("bad key = %q, must be untouched", v) } if ttl := w.TTL("k1"); ttl <= 30*24*time.Hour { t.Errorf("TTL(k1) = %v, EXPIRE should still have run", ttl) } } // GetMaster reads from the write side (used after a lost reply, when the // replica may not have the increment yet). func TestGetMaster(t *testing.T) { s, w, r := testStore(t) w.Set("k", "19") r.Set("k", "3") got, err := s.GetMaster(context.Background(), []string{"k", "missing"}) if err != nil || len(got) != 2 || got[0] != 19 || got[1] != 0 { t.Errorf("GetMaster = %v, %v; want [19 0] from the master", got, err) } } // OutcomeUnknown tells whether a failed Incr may still have been applied: a // refused connection never reached Redis, a timeout or reset after the write // might have. func TestOutcomeUnknown(t *testing.T) { s, w, _ := testStore(t) w.Close() _, err := s.Incr(context.Background(), []string{"k"}) if err == nil { t.Fatal("expected an error with the write side down") } if OutcomeUnknown(err) { t.Errorf("connection refused must be a known failure, got unknown: %v", err) } if OutcomeUnknown(nil) || OutcomeUnknown(context.Canceled) { t.Error("nil / canceled are not unknown outcomes") } for _, e := range []error{context.DeadlineExceeded, io.EOF, &net.OpError{Op: "read", Err: timeoutErr{}}, errors.New("i/o timeout")} { if !OutcomeUnknown(e) { t.Errorf("%v should be an unknown outcome", e) } } } type timeoutErr struct{} func (timeoutErr) Error() string { return "i/o timeout" } func (timeoutErr) Timeout() bool { return true } func (timeoutErr) Temporary() bool { return true } // Fifty concurrent increments of one key must end at exactly 50 — the // pipeline relies on INCRBY being atomic, not on read-then-write. func TestIncr_Concurrent(t *testing.T) { s, w, _ := testStore(t) var wg sync.WaitGroup for i := 0; i < 50; i++ { wg.Add(1) go func() { defer wg.Done() if _, err := s.Incr(context.Background(), []string{"k"}); err != nil { t.Error(err) } }() } wg.Wait() if v, _ := w.Get("k"); v != "50" { t.Errorf("k = %q, want 50", v) } } // Get reads from the replica side: a key present only on the write instance // is invisible, a key present on the read instance is returned, missing = 0. func TestGet_UsesReadSide(t *testing.T) { s, w, r := testStore(t) w.Set("only-on-write", "9") r.Set("on-read", "5") got, err := s.Get(context.Background(), []string{"on-read", "only-on-write", "missing"}) if err != nil { t.Fatal(err) } if len(got) != 3 || got[0] != 5 || got[1] != 0 || got[2] != 0 { t.Errorf("Get = %v, want [5 0 0]", got) } if got, err := s.Get(context.Background(), nil); err != nil || len(got) != 0 { t.Errorf("Get(no keys) = %v, %v", got, err) } } func TestGet_RoundRobinReaders(t *testing.T) { w, r1, r2 := newMini(t), newMini(t), newMini(t) r1.Set("k", "1") r2.Set("k", "2") s := NewWithClients(redis.NewClient(&redis.Options{Addr: w.Addr()}), []*redis.Client{ redis.NewClient(&redis.Options{Addr: r1.Addr()}), redis.NewClient(&redis.Options{Addr: r2.Addr()}), }) defer s.Close() seen := map[int64]int{} for i := 0; i < 4; i++ { got, err := s.Get(context.Background(), []string{"k"}) if err != nil { t.Fatal(err) } seen[got[0]]++ } if seen[1] != 2 || seen[2] != 2 { t.Errorf("reads not alternated between replicas: %v", seen) } } func TestErrors_RedisDown(t *testing.T) { s, w, r := testStore(t) ctx := context.Background() if err := s.Ping(ctx); err != nil { t.Fatalf("Ping before close: %v", err) } w.Close() if _, err := s.Incr(ctx, []string{"k"}); err == nil { t.Error("Incr with write side down should fail") } if err := s.Ping(ctx); err == nil { t.Error("Ping with write side down should fail") } r.Close() if _, err := s.Get(ctx, []string{"k"}); err == nil { t.Error("Get with read side down should fail") } } func TestNew_Static(t *testing.T) { w, r := newMini(t), newMini(t) s, err := New(config.RedisConfig{WriteAddr: w.Addr(), ReadAddrs: []string{r.Addr()}, PoolSize: 2, TimeoutMS: 200}) if err != nil { t.Fatalf("New: %v", err) } defer s.Close() ctx := context.Background() if err := s.Ping(ctx); err != nil { t.Fatalf("Ping: %v", err) } if _, err := s.Incr(ctx, []string{"k"}); err != nil { t.Fatal(err) } if v, _ := w.Get("k"); v != "1" { t.Errorf("write went to %q, want the write_addr instance", v) } r.Set("k", "7") if got, _ := s.Get(ctx, []string{"k"}); got[0] != 7 { t.Errorf("Get = %v, want 7 from read_addrs", got) } } func TestNew_StaticDefaultsReadToWrite(t *testing.T) { w := newMini(t) s, err := New(config.RedisConfig{WriteAddr: w.Addr(), PoolSize: 1, TimeoutMS: 200}) if err != nil { t.Fatal(err) } defer s.Close() w.Set("k", "3") if got, err := s.Get(context.Background(), []string{"k"}); err != nil || got[0] != 3 { t.Errorf("Get = %v, %v; want 3 from the write instance", got, err) } } func TestNew_RejectsNonZeroDB(t *testing.T) { if _, err := New(config.RedisConfig{WriteAddr: "127.0.0.1:6379", DB: 1, PoolSize: 1, TimeoutMS: 100}); err == nil { t.Error("New with db=1 should fail: migadu_web reads DB 0") } } // Sentinel mode constructs lazily; an unreachable sentinel surfaces on the // first command, bounded by the configured timeout. func TestNew_SentinelUnreachable(t *testing.T) { s, err := New(config.RedisConfig{MasterName: "mymaster", Sentinels: []string{"127.0.0.1:1"}, PoolSize: 1, TimeoutMS: 200}) if err != nil { t.Fatalf("New: %v", err) } defer s.Close() ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() start := time.Now() if err := s.Ping(ctx); err == nil { t.Error("Ping against an unreachable sentinel should fail") } if time.Since(start) > 2500*time.Millisecond { t.Errorf("Ping took %v, should fail fast", time.Since(start)) } } func TestClose(t *testing.T) { s, _, _ := testStore(t) if err := s.Close(); err != nil { t.Fatalf("Close: %v", err) } if err := s.Ping(context.Background()); err == nil { t.Error("Ping after Close should fail") } }