Eric Bower
·
2026-09-10
1package main
2
3import (
4 "bufio"
5 "encoding/json"
6 "errors"
7 "flag"
8 "fmt"
9 "io"
10 "io/fs"
11 "log/slog"
12 "net"
13 "os"
14 "path"
15 "path/filepath"
16 "strconv"
17 "strings"
18 "time"
19
20 "github.com/picosh/pico/pkg/shared"
21)
22
23// StorageData matches caddy-storage-redis serialization schema.
24type StorageData struct {
25 Value []byte `json:"value"`
26 Modified time.Time `json:"modified"`
27 Size int64 `json:"size"`
28 Compression int `json:"compression"`
29 Encryption int `json:"encryption"`
30}
31
32// ValkeyClient is a lightweight RESP client using only standard library.
33type ValkeyClient struct {
34 conn net.Conn
35 br *bufio.Reader
36 bw *bufio.Writer
37}
38
39func NewValkeyClient(addr, password string, db int, timeout time.Duration) (*ValkeyClient, error) {
40 conn, err := net.DialTimeout("tcp", addr, timeout)
41 if err != nil {
42 return nil, fmt.Errorf("connect to valkey (%s): %w", addr, err)
43 }
44
45 client := &ValkeyClient{
46 conn: conn,
47 br: bufio.NewReader(conn),
48 bw: bufio.NewWriter(conn),
49 }
50
51 if password != "" {
52 if _, err := client.Do("AUTH", password); err != nil {
53 _ = conn.Close()
54 return nil, fmt.Errorf("authentication failed: %w", err)
55 }
56 }
57
58 if db > 0 {
59 if _, err := client.Do("SELECT", strconv.Itoa(db)); err != nil {
60 _ = conn.Close()
61 return nil, fmt.Errorf("select db %d failed: %w", db, err)
62 }
63 }
64
65 return client, nil
66}
67
68func (c *ValkeyClient) Close() error {
69 return c.conn.Close()
70}
71
72func (c *ValkeyClient) Do(args ...string) (any, error) {
73 if err := c.conn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil {
74 return nil, err
75 }
76
77 if _, err := fmt.Fprintf(c.bw, "*%d\r\n", len(args)); err != nil {
78 return nil, err
79 }
80 for _, arg := range args {
81 if _, err := fmt.Fprintf(c.bw, "$%d\r\n%s\r\n", len(arg), arg); err != nil {
82 return nil, err
83 }
84 }
85 if err := c.bw.Flush(); err != nil {
86 return nil, err
87 }
88
89 return c.readReply()
90}
91
92func (c *ValkeyClient) readReply() (any, error) {
93 line, err := c.br.ReadString('\n')
94 if err != nil {
95 return nil, err
96 }
97 if len(line) < 3 {
98 return nil, fmt.Errorf("malformed RESP reply: %q", line)
99 }
100
101 prefix := line[0]
102 payload := strings.TrimRight(line[1:], "\r\n")
103
104 switch prefix {
105 case '+': // Simple string
106 return payload, nil
107 case '-': // Error
108 return nil, errors.New(payload)
109 case ':': // Integer
110 return strconv.ParseInt(payload, 10, 64)
111 case '$': // Bulk string
112 length, err := strconv.Atoi(payload)
113 if err != nil {
114 return nil, err
115 }
116 if length == -1 {
117 return nil, nil // Nil
118 }
119 buf := make([]byte, length+2)
120 if _, err := io.ReadFull(c.br, buf); err != nil {
121 return nil, err
122 }
123 return string(buf[:length]), nil
124 case '*': // Array
125 count, err := strconv.Atoi(payload)
126 if err != nil {
127 return nil, err
128 }
129 if count == -1 {
130 return nil, nil
131 }
132 items := make([]any, count)
133 for i := 0; i < count; i++ {
134 items[i], err = c.readReply()
135 if err != nil {
136 return nil, err
137 }
138 }
139 return items, nil
140 default:
141 return nil, fmt.Errorf("unknown RESP prefix: %c", prefix)
142 }
143}
144
145func (c *ValkeyClient) Exists(key string) (bool, error) {
146 res, err := c.Do("EXISTS", key)
147 if err != nil {
148 return false, err
149 }
150 if n, ok := res.(int64); ok {
151 return n > 0, nil
152 }
153 return false, nil
154}
155
156func (c *ValkeyClient) Set(key, value string) error {
157 res, err := c.Do("SET", key, value)
158 if err != nil {
159 return err
160 }
161 if s, ok := res.(string); ok && s == "OK" {
162 return nil
163 }
164 return fmt.Errorf("unexpected set response: %v", res)
165}
166
167func (c *ValkeyClient) ZAdd(key string, score float64, member string) error {
168 _, err := c.Do("ZADD", key, strconv.FormatFloat(score, 'f', -1, 64), member)
169 return err
170}
171
172// splitDirectoryKey reproduces caddy-storage-redis's directory hierarchy split.
173func splitDirectoryKey(key string, baseIsDir bool) (string, string) {
174 dir := path.Dir(key)
175 base := path.Base(key)
176 if baseIsDir {
177 base += "/"
178 }
179 return dir, base
180}
181
182// storeDirectoryRecord recursively adds records into Redis Sorted Sets matching caddy-storage-redis.
183func storeDirectoryRecord(client *ValkeyClient, key string, score float64, baseIsDir bool) error {
184 dir, base := splitDirectoryKey(key, baseIsDir)
185 if dir == "." {
186 return nil
187 }
188
189 if err := client.ZAdd(dir, score, base); err != nil {
190 return fmt.Errorf("unable to add %s to set %s: %w", base, dir, err)
191 }
192
193 return storeDirectoryRecord(client, dir, score, true)
194}
195
196type syncStats struct {
197 found int
198 synced int
199 skipped int
200 failed int
201}
202
203// findCaddyRoots detects potential Caddy storage roots in given path.
204func findCaddyRoots(targetDir string) []string {
205 var roots []string
206
207 // Check if targetDir is itself a Caddy storage root
208 if isCaddyStorageRoot(targetDir) {
209 return []string{targetDir}
210 }
211
212 // Check immediate subdirectories (e.g. `data/caddy` or service dirs)
213 entries, err := os.ReadDir(targetDir)
214 if err != nil {
215 return nil
216 }
217
218 for _, e := range entries {
219 if !e.IsDir() || e.Name() == "archive" {
220 continue
221 }
222 sub := filepath.Join(targetDir, e.Name())
223 if isCaddyStorageRoot(sub) {
224 roots = append(roots, sub)
225 continue
226 }
227
228 // Also check nested `.../data/caddy` commonly used in compose mounts
229 nested := filepath.Join(sub, "data", "caddy")
230 if isCaddyStorageRoot(nested) {
231 roots = append(roots, nested)
232 continue
233 }
234 nestedCaddy := filepath.Join(sub, "caddy")
235 if isCaddyStorageRoot(nestedCaddy) {
236 roots = append(roots, nestedCaddy)
237 }
238 }
239
240 return roots
241}
242
243func isCaddyStorageRoot(dir string) bool {
244 certs := filepath.Join(dir, "certificates")
245 acme := filepath.Join(dir, "acme")
246 if fi, err := os.Stat(certs); err == nil && fi.IsDir() {
247 return true
248 }
249 if fi, err := os.Stat(acme); err == nil && fi.IsDir() {
250 return true
251 }
252 return false
253}
254
255func syncFile(
256 client *ValkeyClient,
257 root string,
258 filePath string,
259 keyPrefix string,
260 dryRun bool,
261 overwrite bool,
262 logger *slog.Logger,
263) (synced bool, skipped bool, err error) {
264 rel, err := filepath.Rel(root, filePath)
265 if err != nil {
266 return false, false, err
267 }
268
269 // Normalize to slash path
270 caddyKey := filepath.ToSlash(rel)
271
272 // Only sync active Caddy storage assets: certificates, acme accounts, and ocsp staples
273 if !strings.HasPrefix(caddyKey, "certificates/") &&
274 !strings.HasPrefix(caddyKey, "acme/") &&
275 !strings.HasPrefix(caddyKey, "ocsp/") {
276 return false, true, nil
277 }
278
279 // Skip temporary locks, write tests, or hidden files
280 baseName := filepath.Base(caddyKey)
281 if strings.Contains(caddyKey, "/locks/") || strings.HasPrefix(baseName, ".") || strings.HasPrefix(baseName, "rw_test_") {
282 return false, true, nil
283 }
284
285 prefixedKey := path.Join(keyPrefix, caddyKey)
286
287 fi, err := os.Stat(filePath)
288 if err != nil {
289 return false, false, err
290 }
291
292 if !overwrite && client != nil {
293 exists, err := client.Exists(prefixedKey)
294 if err != nil {
295 return false, false, fmt.Errorf("check exists %s: %w", prefixedKey, err)
296 }
297 if exists {
298 logger.Debug("key already exists in valkey, skipping", "key", prefixedKey)
299 return false, true, nil
300 }
301 }
302
303 content, err := os.ReadFile(filePath)
304 if err != nil {
305 return false, false, err
306 }
307
308 sd := StorageData{
309 Value: content,
310 Modified: fi.ModTime().UTC(),
311 Size: int64(len(content)),
312 Compression: 0,
313 Encryption: 0,
314 }
315
316 jsonBytes, err := json.Marshal(sd)
317 if err != nil {
318 return false, false, fmt.Errorf("marshal storage data for %s: %w", caddyKey, err)
319 }
320
321 if dryRun {
322 logger.Info("[dry-run] would sync", "key", prefixedKey, "size", len(content), "modTime", sd.Modified)
323 return true, false, nil
324 }
325
326 score := float64(sd.Modified.Unix())
327 if err := storeDirectoryRecord(client, prefixedKey, score, false); err != nil {
328 return false, false, fmt.Errorf("index directory for %s: %w", prefixedKey, err)
329 }
330
331 if err := client.Set(prefixedKey, string(jsonBytes)); err != nil {
332 return false, false, fmt.Errorf("set key %s: %w", prefixedKey, err)
333 }
334
335 logger.Info("synced", "key", prefixedKey, "size", len(content))
336 return true, false, nil
337}
338
339func main() {
340 defaultAddr := shared.GetEnv("VALKEY_ADDR", shared.GetEnv("REDIS_ADDR", ""))
341 defaultPass := shared.GetEnv("VALKEY_PASSWORD", shared.GetEnv("REDIS_PASSWORD", ""))
342 defaultPrefix := shared.GetEnv("KEY_PREFIX", "caddy")
343
344 addrFlag := flag.String("addr", defaultAddr, "Valkey/Redis server address (host:port)")
345 passFlag := flag.String("password", defaultPass, "Valkey/Redis password")
346 dbFlag := flag.Int("db", 0, "Valkey database index")
347 prefixFlag := flag.String("prefix", defaultPrefix, "Caddy storage key prefix")
348 dirFlag := flag.String("dir", "", "Directory containing Caddy storage (required)")
349 dryRunFlag := flag.Bool("dry-run", false, "Preview keys without uploading to Valkey")
350 overwriteFlag := flag.Bool("overwrite", false, "Overwrite keys even if they already exist in Valkey")
351 verboseFlag := flag.Bool("verbose", false, "Enable verbose logging")
352 flag.Parse()
353
354 logLevel := slog.LevelInfo
355 if *verboseFlag {
356 logLevel = slog.LevelDebug
357 }
358 logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: logLevel}))
359
360 dir := *dirFlag
361 if dir == "" && flag.NArg() > 0 {
362 dir = flag.Arg(0)
363 }
364 if dir == "" {
365 logger.Error("directory is required (provide -dir <path> or path as argument)")
366 os.Exit(1)
367 }
368
369 if *addrFlag == "" {
370 logger.Error("address is required (provide -addr or set VALKEY_ADDR)")
371 os.Exit(1)
372 }
373
374 if *passFlag == "" {
375 logger.Error("password is required (provide -password or set VALKEY_PASSWORD)")
376 os.Exit(1)
377 }
378
379 roots := findCaddyRoots(dir)
380 if len(roots) == 0 {
381 logger.Error("no Caddy storage directories found in path", "dir", dir)
382 os.Exit(1)
383 }
384
385 logger.Info("discovered Caddy storage roots", "count", len(roots), "roots", roots)
386
387 var client *ValkeyClient
388 if !*dryRunFlag {
389 var err error
390 client, err = NewValkeyClient(*addrFlag, *passFlag, *dbFlag, 5*time.Second)
391 if err != nil {
392 logger.Error("failed to connect to Valkey", "err", err, "addr", *addrFlag)
393 os.Exit(1)
394 }
395 defer func() { _ = client.Close() }()
396 logger.Info("connected to Valkey successfully", "addr", *addrFlag, "db", *dbFlag)
397 } else {
398 logger.Info("running in DRY-RUN mode; no changes will be made to Valkey")
399 }
400
401 stats := syncStats{}
402
403 for _, root := range roots {
404 logger.Info("syncing Caddy root", "path", root)
405 err := filepath.WalkDir(root, func(p string, d fs.DirEntry, walkErr error) error {
406 if walkErr != nil {
407 return walkErr
408 }
409 if d.IsDir() {
410 return nil
411 }
412
413 stats.found++
414 synced, skipped, err := syncFile(client, root, p, *prefixFlag, *dryRunFlag, *overwriteFlag, logger)
415 if err != nil {
416 logger.Error("failed to sync file", "path", p, "err", err)
417 stats.failed++
418 return nil
419 }
420 if synced {
421 stats.synced++
422 } else if skipped {
423 stats.skipped++
424 }
425 return nil
426 })
427 if err != nil {
428 logger.Error("error walking directory", "root", root, "err", err)
429 }
430 }
431
432 fmt.Printf("\n--- Valkey TLS Migration Summary ---\n")
433 fmt.Printf("Total files found: %d\n", stats.found)
434 fmt.Printf("Successfully synced: %d\n", stats.synced)
435 fmt.Printf("Skipped (existing): %d\n", stats.skipped)
436 fmt.Printf("Failed: %d\n", stats.failed)
437 if *dryRunFlag {
438 fmt.Printf("Mode: DRY RUN (no keys written)\n")
439 }
440 fmt.Printf("------------------------------------\n")
441
442 if stats.failed > 0 {
443 os.Exit(1)
444 }
445}