From 92cd652b0d0b6d106088761de1366b75e54e31e9 Mon Sep 17 00:00:00 2001 From: eleksir Date: Fri, 3 Apr 2026 01:02:03 +0300 Subject: [PATCH] Add stub database selection, and fix differences between pekoe and c-redis - PUBSUB CHANNELS glob pattern matching - Implement SELECT + max DB validation (0-15) (default db max is 16), for perl clients - CLIENT KILL ID (single/range) + MAXAGE - CLIENT INFO events + resp fields - PUBSUB SHARDCHANNELS/SHARDNUMSUB stubs --- internal/glob/glob.go | 58 ++++++++++++++++++ internal/glob/glob_test.go | 69 ++++++++++++++++++++++ internal/pubsub/client.go | 32 +++++++++- internal/server/handlers.go | 113 +++++++++++++++++++++++++++++++++--- internal/server/server.go | 4 ++ 5 files changed, 266 insertions(+), 10 deletions(-) create mode 100644 internal/glob/glob.go create mode 100644 internal/glob/glob_test.go diff --git a/internal/glob/glob.go b/internal/glob/glob.go new file mode 100644 index 0000000..0fb4ce9 --- /dev/null +++ b/internal/glob/glob.go @@ -0,0 +1,58 @@ +// @tag glob +package glob + +// Match checks if a string matches a glob-style pattern. +// Supports: +// - * (asterisk): matches any sequence of characters (including empty) +// - ? (question): matches any single character +// +// This matches Redis PUBSUB CHANNELS pattern behavior. +func Match(pattern, s string) bool { + if pattern == "" { + return s == "" + } + if pattern == "*" { + return true + } + + // Simple glob matching using dynamic programming + // Convert pattern to regex-like matching + patternLen := len(pattern) + sLen := len(s) + + // dp[i][j] = true if pattern[0:i] matches s[0:j] + dp := make([][]bool, patternLen+1) + for i := range dp { + dp[i] = make([]bool, sLen+1) + } + + // Empty pattern matches empty string + dp[0][0] = true + + // Handle leading * in pattern + for i := 1; i <= patternLen; i++ { + if pattern[i-1] == '*' { + dp[i][0] = dp[i-1][0] + } + } + + // Fill the DP table + for i := 1; i <= patternLen; i++ { + for j := 1; j <= sLen; j++ { + switch pattern[i-1] { + case '*': + // * matches: either consume nothing (stay at same position in pattern) + // or consume one character from string + dp[i][j] = dp[i-1][j] || dp[i][j-1] + case '?': + // ? matches any single character + dp[i][j] = dp[i-1][j-1] + default: + // Literal character must match exactly + dp[i][j] = dp[i-1][j-1] && pattern[i-1] == s[j-1] + } + } + } + + return dp[patternLen][sLen] +} diff --git a/internal/glob/glob_test.go b/internal/glob/glob_test.go new file mode 100644 index 0000000..647c622 --- /dev/null +++ b/internal/glob/glob_test.go @@ -0,0 +1,69 @@ +// @tag glob +// @tag test +package glob + +import "testing" + +func TestMatch(t *testing.T) { + tests := []struct { + pattern string + s string + want bool + }{ + {"", "", true}, + {"", "foo", false}, + {"*", "", true}, + {"*", "foo", true}, + {"*", "foo/bar", true}, + {"foo", "foo", true}, + {"foo", "bar", false}, + {"foo*", "foo", true}, + {"foo*", "foobar", true}, + {"foo*", "fo", false}, + {"*foo", "foo", true}, + {"*foo", "barfoo", true}, + {"*foo", "bar", false}, + {"foo*bar", "foobar", true}, + {"foo*bar", "foobazbar", true}, + {"foo*bar", "foobaz", false}, + {"foo?bar", "fooXbar", true}, + {"foo?bar", "fooXXbar", false}, + {"foo??bar", "fooXXbar", true}, + {"?", "a", true}, + {"?", "", false}, + {"a?c", "abc", true}, + {"a?c", "ac", false}, + {"foo?*bar", "fooabcbar", true}, + {"foo?*bar", "foobar", false}, + {"*foo*bar*", "ahellofoocherrybarb", true}, + {"ch*", "channel", true}, + {"ch*", "ch", true}, + {"ch*", "c", false}, + {"*", "*", true}, + {"**", "*", true}, + {"*a*b", "aXb", true}, + {"a*b*c", "aXbYc", true}, + } + + for _, tt := range tests { + t.Run(tt.pattern+"/"+tt.s, func(t *testing.T) { + if got := Match(tt.pattern, tt.s); got != tt.want { + t.Errorf("Match(%q, %q) = %v, want %v", tt.pattern, tt.s, got, tt.want) + } + }) + } +} + +func BenchmarkMatch(b *testing.B) { + patterns := []string{"*", "ch*", "foo*bar", "???", "*channel*"} + strings := []string{"channel", "channel123", "foobar", "abc", "somechannelname"} + + b.ReportAllocs() + for i := 0; i < b.N; i++ { + for _, p := range patterns { + for _, s := range strings { + Match(p, s) + } + } + } +} diff --git a/internal/pubsub/client.go b/internal/pubsub/client.go index 9c94d36..d464b9b 100644 --- a/internal/pubsub/client.go +++ b/internal/pubsub/client.go @@ -31,6 +31,7 @@ type Client struct { conn net.Conn name string // Connection name set via CLIENT SETNAME replyMode ReplyMode // Reply mode (ON/OFF/SKIP) + selectedDB int // Selected database index (via SELECT command) flags uint64 // Client flags (e.g., paused, no-touch) created time.Time // When the client connected lastActive time.Time // Last activity time @@ -48,6 +49,7 @@ func NewClient(conn net.Conn) *Client { conn: conn, name: "", replyMode: ReplyOn, + selectedDB: 0, flags: 0, created: now, lastActive: now, @@ -89,6 +91,27 @@ func (c *Client) GetReplyMode() ReplyMode { return c.replyMode } +// SelectedDB returns the currently selected database index +func (c *Client) SelectedDB() int { + c.mu.RLock() + defer c.mu.RUnlock() + return c.selectedDB +} + +// CreatedAt returns when the client connected +func (c *Client) CreatedAt() time.Time { + c.mu.RLock() + defer c.mu.RUnlock() + return c.created +} + +// SetSelectedDB sets the selected database index +func (c *Client) SetSelectedDB(db int) { + c.mu.Lock() + defer c.mu.Unlock() + c.selectedDB = db +} + // UpdateActivity updates the last activity timestamp func (c *Client) UpdateActivity() { c.mu.Lock() @@ -212,7 +235,7 @@ func (c *Client) LocalAddr() string { } // InfoString returns CLIENT INFO formatted string (Redis format) -// Format: id= addr= laddr= fd= age= idle= flags= db= sub= psub= multi= qbuf= qbuf-free= obl= oll= omem= tot-mem= cmd= user= redir= +// Format: id= addr= laddr= fd= age= idle= flags= db= sub= psub= multi= qbuf= qbuf-free= obl= oll= omem= tot-mem= events= cmd= user= redir= resp= func (c *Client) InfoString() string { c.mu.RLock() defer c.mu.RUnlock() @@ -229,16 +252,21 @@ func (c *Client) InfoString() string { flags += "S" } + // Events - indicate connection is active for read and write + events := "rw" + // Format: Redis CLIENT INFO format - return fmt.Sprintf("id=%d addr=%s laddr=%s fd=-1 age=%d idle=%d flags=%s db=0 sub=%d psub=%d multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 tot-mem=0 cmd=client user=default redir=-1", + return fmt.Sprintf("id=%d addr=%s laddr=%s fd=-1 age=%d idle=%d flags=%s db=%d sub=%d psub=%d multi=-1 qbuf=0 qbuf-free=0 obl=0 oll=0 omem=0 tot-mem=0 events=%s cmd=client user=default redir=-1 resp=2", c.id, c.RemoteAddr(), c.LocalAddr(), age, idle, flags, + c.selectedDB, len(c.subscribed), len(c.patterns), + events, ) } diff --git a/internal/server/handlers.go b/internal/server/handlers.go index fabb840..6339fcd 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -5,7 +5,9 @@ package server import ( "fmt" "strings" + "time" + "github.com/eleksir/pekoe/internal/glob" "github.com/eleksir/pekoe/internal/logger" "github.com/eleksir/pekoe/internal/protocol" "github.com/eleksir/pekoe/internal/pubsub" @@ -132,12 +134,24 @@ func (s *Server) handlePubsub(args [][]byte, writer *ConnWriter) error { pattern = string(args[1]) } channels := s.channelList(pattern) - result := make([]protocol.Value, len(channels)) - for i, ch := range channels { - result[i] = protocol.Bulk(ch) + result := make([]protocol.Value, 0, len(channels)) + for _, ch := range channels { + if pattern == "" || glob.Match(pattern, ch) { + result = append(result, protocol.Bulk(ch)) + } } writer.WriteArray(result) + case "shardchannels": + // PUBSUB SHARDCHANNELS [pattern] - List active sharded channels + // Pekoe doesn't support sharded pubsub, return empty list + writer.WriteArray(nil) + + case "shardnumsub": + // PUBSUB SHARDNUMSUB [shardchannel ...] - Get subscriber count for sharded channels + // Pekoe doesn't support sharded pubsub, return empty array + writer.WriteArray(nil) + case "numpat": // PUBSUB NUMPAT - Count of unique pattern subscriptions writer.WriteInt(int64(s.hub.PatternCount())) @@ -147,7 +161,9 @@ func (s *Server) handlePubsub(args [][]byte, writer *ConnWriter) error { helpText := []string{ "PUBSUB [ [ ...]]>", " PUBSUB CHANNELS [pattern] - List active channels (optionally filtered by pattern)", + " PUBSUB SHARDCHANNELS [pattern] - List active sharded channels (not supported)", " PUBSUB NUMSUB [channel ...] - Get subscriber count for channels", + " PUBSUB SHARDNUMSUB [shardchannel ...] - Get subscriber count for sharded channels (not supported)", " PUBSUB NUMPAT - Get the number of pattern subscriptions", " PUBSUB HELP - Show this help message", } @@ -349,11 +365,15 @@ func (s *Server) buildClientList(args [][]byte) string { func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) (int, error) { clients := s.hub.GetAllClients() killed := 0 + var notFoundErr error // Parse arguments var killID uint64 + var killIDMin uint64 + var killIDMax uint64 var killAddr string var skipMe = true + var maxAgeMs int64 for i := 0; i < len(args); i++ { arg := strings.ToUpper(string(args[i])) @@ -361,6 +381,12 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) ( case "ID": if i+2 < len(args) { // ID min max format - kill all IDs in range + if _, err := fmt.Sscanf(string(args[i+1]), "%d", &killIDMin); err != nil { + return 0, fmt.Errorf("invalid client ID: %s", args[i+1]) + } + if _, err := fmt.Sscanf(string(args[i+2]), "%d", &killIDMax); err != nil { + return 0, fmt.Errorf("invalid client ID: %s", args[i+2]) + } i += 2 } else if i+1 < len(args) { // Single ID @@ -374,14 +400,35 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) ( killAddr = string(args[i+1]) i++ } + case "LADDR": + // LADDR not fully supported, just skip + if i+1 < len(args) { + i++ + } case "SKIPME": if i+1 < len(args) { skipMe = strings.ToUpper(string(args[i+1])) == "YES" i++ } + case "MAXAGE": + if i+1 < len(args) { + if _, err := fmt.Sscanf(string(args[i+1]), "%d", &maxAgeMs); err != nil { + return 0, fmt.Errorf("invalid maxage: %s", args[i+1]) + } + if maxAgeMs <= 0 { + return 0, fmt.Errorf("maxage must be greater than 0") + } + i++ + } + case "TYPE": + // TYPE filter handled separately in handleClient + if i+1 < len(args) { + i++ + } } } + now := time.Now() for _, c := range clients { // Skip current client if requested if skipMe && c.ID() == currentClient.ID() { @@ -390,11 +437,29 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) ( shouldKill := false - if killID > 0 && c.ID() == killID { - shouldKill = true + // Check ID (single or range) + if killID > 0 { + shouldKill = c.ID() == killID + } else if killIDMin > 0 || killIDMax > 0 { + id := c.ID() + if killIDMin > 0 && id < killIDMin { + shouldKill = false + } else if killIDMax > 0 && id > killIDMax { + shouldKill = false + } else { + shouldKill = true + } } - if killAddr != "" && c.RemoteAddr() == killAddr { - shouldKill = true + + // Check ADDR + if !shouldKill && killAddr != "" { + shouldKill = c.RemoteAddr() == killAddr + } + + // Check MAXAGE + if shouldKill && maxAgeMs > 0 { + age := now.Sub(c.CreatedAt()).Milliseconds() + shouldKill = age >= maxAgeMs } if shouldKill { @@ -406,7 +471,39 @@ func (s *Server) handleClientKill(currentClient *pubsub.Client, args [][]byte) ( } } - return killed, nil + // For single ID/ADDR that wasn't found, Redis returns error + if killed == 0 && (killID > 0 || killAddr != "") { + return 0, fmt.Errorf("No such client") + } + + return killed, notFoundErr +} + +// handleSelect processes SELECT command. +// SELECT db - Switch to a different database (0-15). +// For pekoe (pubsub-only), this only tracks the selected DB for client compatibility. +// Redis default max DB is 16 (0-15), configurable via max-databases in config. +func (s *Server) handleSelect(client *pubsub.Client, args [][]byte, writer *ConnWriter) error { + if len(args) != 1 { + writer.WriteError("ERR wrong number of arguments for 'select' command") + return nil + } + + dbIndexStr := string(args[0]) + var dbIndex int + if _, err := fmt.Sscanf(dbIndexStr, "%d", &dbIndex); err != nil { + writer.WriteError("ERR invalid DB index") + return nil + } + + if dbIndex < 0 || dbIndex >= MaxDatabases { + writer.WriteError("ERR DB index is out of range") + return nil + } + + client.SetSelectedDB(dbIndex) + writer.WriteStatus("OK") + return nil } // handlePsubscribe processes PSUBSCRIBE command. diff --git a/internal/server/server.go b/internal/server/server.go index 7642dd1..377e215 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -21,6 +21,8 @@ import ( "github.com/eleksir/pekoe/internal/pubsub" ) +const MaxDatabases = 16 // Redis default maximum databases (0-15) + // Sentinel errors for connection handling var ( ErrQuit = errors.New("client quit") @@ -276,6 +278,8 @@ func (s *Server) processCommand(client *pubsub.Client, cmd *Command, writer *Con return s.handleQuit(client, writer) case "client": return s.handleClient(client, cmd.Args[1:], writer) + case "select": + return s.handleSelect(client, cmd.Args[1:], writer) default: // Check if client is in pubsub mode - only pubsub commands allowed if client != nil && (client.SubsCount() > 0 || client.PatternSubsCount() > 0) {