Notifications: part 1

.github/workflows/go.yml

@@ -25,7 +25,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v6
       with:
-        go-version: '1.25.1'
+        go-version: '1.25.2'
         cache: true
 
     - name: Build
@@ -43,7 +43,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v6
       with:
-        go-version: '1.25.1'
+        go-version: '1.25.2'
         cache: true
 
     - name: Build Caddy
@@ -71,7 +71,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v6
       with:
-        go-version: '1.25.1'
+        go-version: '1.25.2'
         cache: true
 
     - name: Build

atlas/caddy/module/module.go

@@ -123,7 +123,6 @@ func (m *Module) Provision(ctx caddy.Context) (err error) {
 }
 
 func (m *Module) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyhttp.Handler) error {
-	fmt.Println("ServeHTTP called")
 	if r.ProtoMajor == 2 && r.Header.Get("content-type") == "application/grpc" {
 		// check authorization
 		authHeader := r.Header.Get("Authorization")
@@ -173,6 +172,10 @@ func (m *Module) UnmarshalCaddyfile(d *caddyfile.Dispenser) (err error) {
 					err = nil
 				}
 
+				if !strings.HasSuffix(path, "/") {
+					path = path + "/"
+				}
+
 				options.CurrentOptions.DbFilename = path + options.CurrentOptions.DbFilename
 				options.CurrentOptions.MetaFilename = path + options.CurrentOptions.MetaFilename
 			case "region":
@@ -273,7 +276,7 @@ func init() {
 
 			ready:
 
-				options.Logger.Info("🌐 Atlas Client Started")
+				options.Logger.Debug("🌐 Atlas Client Started")
 
 				rl, err := readline.New("> ")
 				if err != nil {

atlas/commands/acl-command.go

@@ -23,7 +23,7 @@ import (
 	"fmt"
 
 	"github.com/bottledcode/atlas-db/atlas"
-	"github.com/bottledcode/atlas-db/atlas/kv"
+	"github.com/bottledcode/atlas-db/atlas/consensus"
 )
 
 type ACLCommand struct {
@@ -58,16 +58,16 @@ func (c *ACLGrantCommand) Execute(ctx context.Context) ([]byte, error) {
 		return nil, err
 	}
 
-	tableKey, _ := c.SelectNormalizedCommand(2)
+	keyName, _ := c.SelectNormalizedCommand(2)
 	principal := c.SelectCommand(3)
 	permsKeyword, _ := c.SelectNormalizedCommand(4)
 	permissions := c.SelectCommand(5)
 
-	if tableKey == "" || principal == "" || permsKeyword != "PERMS" || permissions == "" {
+	if keyName == "" || principal == "" || permsKeyword != "PERMS" || permissions == "" {
 		return nil, fmt.Errorf("ACL GRANT requires format: ACL GRANT <table> <name> PERMS <permissions>")
 	}
 
-	key := kv.FromDottedKey(tableKey)
+	key := consensus.KeyName(keyName)
 
 	switch permissions {
 	case "READ":
@@ -116,7 +116,7 @@ func (c *ACLRevokeCommand) Execute(ctx context.Context) ([]byte, error) {
 		return nil, fmt.Errorf("ACL REVOKE requires format: ACL REVOKE <table> <name> PERMS <permissions>")
 	}
 
-	key := kv.FromDottedKey(tableKey)
+	key := consensus.KeyName(tableKey)
 	switch permissions {
 	case "READ":
 		err := atlas.RevokeReader(ctx, key, principal)

atlas/commands/key-command.go

@@ -25,7 +25,7 @@ import (
 	"time"
 
 	"github.com/bottledcode/atlas-db/atlas"
-	"github.com/bottledcode/atlas-db/atlas/kv"
+	"github.com/bottledcode/atlas-db/atlas/consensus"
 	"github.com/bottledcode/atlas-db/atlas/options"
 	"go.uber.org/zap"
 )
@@ -66,8 +66,8 @@ func (c *KeyCommand) GetNext() (Command, error) {
 	return EmptyCommandString, nil
 }
 
-func (c *KeyCommand) FromKey(key string) *kv.KeyBuilder {
-	return kv.FromDottedKey(key)
+func (c *KeyCommand) FromKey(key string) consensus.KeyName {
+	return consensus.KeyName(key)
 }
 
 type KeyPutCommand struct {

atlas/commands/key-command_test.go

@@ -88,7 +88,7 @@ func TestKeyGet_FromKey_Mapping(t *testing.T) {
 	// Normalized() uppercases tokens, so SelectNormalizedCommand(2) yields "TABLE.ROW"
 	key, _ := kgc.SelectNormalizedCommand(2)
 	builder := kgc.FromKey(key)
-	if got := builder.String(); got != "t:TABLE:r:ROW" {
+	if got := string(builder); got != "TABLE.ROW" {
 		t.Fatalf("unexpected key mapping, got %q", got)
 	}
 }
@@ -105,7 +105,7 @@ func TestKeyGet_FromKey_Mapping_MultiPart(t *testing.T) {
 	}
 	key, _ := kgc.SelectNormalizedCommand(2)
 	builder := kgc.FromKey(key)
-	if got := builder.String(); got != "t:TABLE:r:ROW:ATTR.MORE" {
+	if got := string(builder); got != "TABLE.ROW.ATTR.MORE" {
 		t.Fatalf("unexpected key mapping, got %q", got)
 	}
 }
@@ -122,12 +122,12 @@ func TestKeyDel_FromKey_Mapping(t *testing.T) {
 	}
 	key, _ := kd.SelectNormalizedCommand(2)
 	builder := kd.FromKey(key)
-	if got := builder.String(); got != "t:TABLE:r:ROW:ATTR.MORE" {
+	if got := string(builder); got != "TABLE.ROW.ATTR.MORE" {
 		t.Fatalf("unexpected key mapping, got %q", got)
 	}
 }
 
-func TestScan_NotImplemented(t *testing.T) {
+func TestScan_ParseCommand(t *testing.T) {
 	cmd := CommandFromString("SCAN prefix")
 	next, err := cmd.GetNext()
 	if err != nil {
@@ -137,8 +137,16 @@ func TestScan_NotImplemented(t *testing.T) {
 	if !ok {
 		t.Fatalf("expected *ScanCommand, got %T", next)
 	}
-	if _, err := sc.Execute(context.Background()); err == nil {
-		t.Fatalf("expected not implemented error for SCAN")
+	// Verify the command structure is correct
+	if err := sc.CheckMinLen(2); err != nil {
+		t.Fatalf("SCAN command should have at least 2 tokens: %v", err)
+	}
+	prefix, ok := sc.SelectNormalizedCommand(1)
+	if !ok {
+		t.Fatalf("expected to select prefix from command")
+	}
+	if prefix != "PREFIX" {
+		t.Fatalf("expected normalized prefix 'PREFIX', got %q", prefix)
 	}
 }
 

atlas/commands/quorum-command.go

@@ -27,7 +27,7 @@ func (q *QuorumInfoCommand) Execute(ctx context.Context) ([]byte, error) {
 	}
 	table, _ := q.SelectNormalizedCommand(2)
 
-	q1, q2, err := consensus.DescribeQuorum(ctx, table)
+	q1, q2, err := consensus.DescribeQuorum(ctx, consensus.KeyName(table))
 	if err != nil {
 		return nil, err
 	}

atlas/commands/scan_commands.go

@@ -4,9 +4,9 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"strings"
 
 	"github.com/bottledcode/atlas-db/atlas"
+	"github.com/bottledcode/atlas-db/atlas/consensus"
 )
 
 type ScanCommand struct{ CommandString }
@@ -22,14 +22,8 @@ func (s *ScanCommand) Execute(ctx context.Context) ([]byte, error) {
 	if !ok {
 		return nil, fmt.Errorf("expected prefix")
 	}
-	parts := strings.Split(prefix, ".")
-	tablePrefix := parts[0]
-	rowPrefix := ""
-	if len(parts) > 1 {
-		rowPrefix = parts[1]
-	}
 
-	keys, err := atlas.PrefixScan(ctx, tablePrefix, rowPrefix)
+	keys, err := atlas.PrefixScan(ctx, consensus.KeyName(prefix))
 	if err != nil {
 		return nil, err
 	}
@@ -41,7 +35,7 @@ func (s *ScanCommand) Execute(ctx context.Context) ([]byte, error) {
 	// Format: KEYS:<count>\n<key1>\n<key2>\n...
 	var buf bytes.Buffer
 	buf.WriteString(fmt.Sprintf("KEYS:%d\n", len(keys)))
-	buf.WriteString(strings.Join(keys, "\n"))
+	buf.Write(bytes.Join(keys, []byte("\n")))
 
 	return buf.Bytes(), nil
 }

atlas/commands/string.go

@@ -79,6 +79,8 @@ func (c *CommandString) GetNext() (Command, error) {
 			return (&QuorumCommand{CommandString: *c}).GetNext()
 		case "ACL":
 			return (&ACLCommand{*c}).GetNext()
+		case "SUB":
+			return (&SubCommand{*c}).GetNext()
 		}
 		return EmptyCommandString, fmt.Errorf("command expected, got %s", next)
 	}

atlas/commands/sub-command.go

@@ -0,0 +1,135 @@
+/*
+ * This file is part of Atlas-DB.
+ *
+ * Atlas-DB is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of
+ * the License, or (at your option) any later version.
+ *
+ * Atlas-DB is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with Atlas-DB. If not, see <https://www.gnu.org/licenses/>.
+ *
+ */
+
+package commands
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"time"
+
+	"github.com/bottledcode/atlas-db/atlas"
+	"github.com/bottledcode/atlas-db/atlas/consensus"
+	"github.com/bottledcode/atlas-db/atlas/options"
+	"go.uber.org/zap"
+)
+
+type SubCommand struct {
+	CommandString
+}
+
+func (c *SubCommand) GetNext() (Command, error) {
+	return c, nil
+}
+
+// SubParsed holds the parsed components of a SUB command.
+type SubParsed struct {
+	Prefix         string
+	URL            string
+	Batch          bool
+	RetryAttempts  int32
+	RetryAfterBase time.Duration
+	Auth           string
+}
+
+// Parse extracts arguments for SUB command.
+// SUB <prefix> <url> [BATCH] [RETRY <attempts>] [RETRY_AFTER <duration>] [AUTH <token>]
+func (c *SubCommand) Parse() (*SubParsed, error) {
+	if err := c.CheckMinLen(3); err != nil {
+		return nil, err
+	}
+
+	prefix, _ := c.SelectNormalizedCommand(1)
+	url := c.SelectCommand(2) // Use raw version to preserve case for URL
+
+	parsed := &SubParsed{
+		Prefix:         prefix,
+		URL:            url,
+		Batch:          true, // Default to non-batched
+		RetryAttempts:  3,    // Default 3 retries
+		RetryAfterBase: 100 * time.Millisecond,
+	}
+
+	// Parse optional flags
+	for i := 3; i < c.NormalizedLen(); i++ {
+		flag, _ := c.SelectNormalizedCommand(i)
+		switch flag {
+		case "NOBATCH":
+			parsed.Batch = false
+		case "RETRY":
+			// Need attempts after RETRY
+			if i+1 >= c.NormalizedLen() {
+				return nil, errors.New("RETRY requires attempts number")
+			}
+			attemptsStr := c.SelectCommand(i + 1)
+			attempts, err := strconv.ParseInt(attemptsStr, 10, 32)
+			if err != nil {
+				return nil, errors.New("RETRY requires valid number")
+			}
+			parsed.RetryAttempts = int32(attempts)
+			i++ // Skip next argument
+		case "RETRY_AFTER":
+			// Need duration after RETRY_AFTER
+			if i+1 >= c.NormalizedLen() {
+				return nil, errors.New("RETRY_AFTER requires duration")
+			}
+			durationStr := c.SelectCommand(i + 1)
+			dur, err := time.ParseDuration(durationStr)
+			if err != nil {
+				return nil, err
+			}
+			parsed.RetryAfterBase = dur
+			i++ // Skip next argument
+		case "AUTH":
+			// Need token after AUTH
+			if i+1 >= c.NormalizedLen() {
+				return nil, errors.New("AUTH requires token")
+			}
+			parsed.Auth = c.SelectCommand(i + 1)
+			i++ // Skip next argument
+		}
+	}
+
+	return parsed, nil
+}
+
+func (c *SubCommand) Execute(ctx context.Context) ([]byte, error) {
+	parsed, err := c.Parse()
+	if err != nil {
+		return nil, err
+	}
+
+	err = atlas.Subscribe(ctx, consensus.KeyName(parsed.Prefix), parsed.URL, atlas.SubscribeOptions{
+		RetryAttempts:  int(parsed.RetryAttempts),
+		RetryAfterBase: parsed.RetryAfterBase,
+		Auth:           parsed.Auth,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	options.Logger.Info("created subscription",
+		zap.String("prefix", parsed.Prefix),
+		zap.String("url", parsed.URL),
+		zap.Bool("batch", parsed.Batch),
+		zap.Int32("retry_attempts", parsed.RetryAttempts),
+		zap.Duration("retry_after_base", parsed.RetryAfterBase))
+
+	return nil, nil
+}