feat(run): add logging, dry-run semantics, health stub

Author: devblac

cmd/watch-tower/run.go

@@ -10,6 +10,7 @@ import (
 	"github.com/devblac/watch-tower/internal/source/algorand"
 	"github.com/devblac/watch-tower/internal/source/evm"
 	"github.com/devblac/watch-tower/internal/storage"
+	"github.com/devblac/watch-tower/internal/logging"
 	"github.com/spf13/cobra"
 )
 
@@ -31,6 +32,7 @@ var runCmd = &cobra.Command{
 	Use:   "run",
 	Short: "Run watch-tower pipelines",
 	RunE: func(cmd *cobra.Command, args []string) error {
+		log := logging.New()
 		ctx := cmd.Context()
 
 		cfg, err := config.Load(cfgPath)
@@ -115,6 +117,7 @@ var runCmd = &cobra.Command{
 			if err := runner.RunOnce(ctx); err != nil {
 				return err
 			}
+			log.Info("tick complete", "dry_run", flagDryRun)
 			if flagOnce {
 				break
 			}

internal/engine/predicate.go

@@ -13,9 +13,10 @@ type Predicate func(args map[string]any) (bool, error)
 // CompilePredicates parses simple expressions into executable predicates.
 // Supported operators: ==, !=, >, <, in, contains.
 // Examples:
-//   "value > 10"
-//   "sender in a,b,c"
-//   "memo contains alert"
+//
+//	"value > 10"
+//	"sender in a,b,c"
+//	"memo contains alert"
 func CompilePredicates(exprs []string) ([]Predicate, error) {
 	var preds []Predicate
 	for _, raw := range exprs {

internal/engine/runner.go

@@ -161,6 +161,10 @@ func (r *Runner) handleEvents(ctx context.Context, events []Event) error {
 		if err != nil || !pass {
 			continue
 		}
+		if r.dryRun {
+			// No side effects in dry-run: skip dedupe and sends.
+			continue
+		}
 		if exec.rule.Dedupe != nil {
 			key := buildDedupeKey(exec.rule.Dedupe.Key, ev)
 			now := r.nowFunc()
@@ -179,9 +183,6 @@ func (r *Runner) handleEvents(ctx context.Context, events []Event) error {
 				return err
 			}
 		}
-		if r.dryRun {
-			continue
-		}
 		for _, sinkID := range exec.rule.Sinks {
 			s := r.sinks[sinkID]
 			if s == nil {

internal/engine/runner_dedupe_test.go

@@ -0,0 +1,24 @@
+package engine
+
+import (
+	"testing"
+)
+
+func TestBuildDedupeKey(t *testing.T) {
+	idx := uint(5)
+	ev := Event{
+		TxHash:   "0xabc",
+		LogIndex: &idx,
+		AppID:    42,
+	}
+
+	key := buildDedupeKey("txhash:logIndex:app_id", ev)
+	if key != "0xabc:5:42" {
+		t.Fatalf("unexpected key: %s", key)
+	}
+
+	key = buildDedupeKey("", ev)
+	if key != "0xabc" {
+		t.Fatalf("default key mismatch: %s", key)
+	}
+}

internal/engine/runner_test.go

@@ -0,0 +1,78 @@
+package engine
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/devblac/watch-tower/internal/config"
+	"github.com/devblac/watch-tower/internal/sink"
+	"github.com/devblac/watch-tower/internal/storage"
+)
+
+type fakeSink struct {
+	count int
+}
+
+func (f *fakeSink) Send(ctx context.Context, payload sink.EventPayload) error {
+	f.count++
+	return nil
+}
+
+// Simple integration: ensure predicates + dedupe + dry-run behave.
+func TestRunnerPredicatesAndDryRun(t *testing.T) {
+	store := newTestStore(t)
+	rule := config.Rule{
+		ID:    "r1",
+		Match: config.MatchSpec{Where: []string{"value > 10"}},
+		Sinks: []string{"s1"},
+		Dedupe: &config.Dedupe{
+			Key: "txhash",
+			TTL: "1h",
+		},
+	}
+	cfg := &config.Config{Rules: []config.Rule{rule}}
+	s := &fakeSink{}
+	runner, err := NewRunner(store, cfg, nil, nil, map[string]sink.Sender{"s1": s}, true, 0, 0)
+	if err != nil {
+		t.Fatalf("runner: %v", err)
+	}
+	runner.nowFunc = func() time.Time { return time.Now() }
+
+	evs := []Event{{
+		RuleID: "r1",
+		TxHash: "0x1",
+		Args:   map[string]any{"value": 20},
+	}}
+	if err := runner.handleEvents(context.Background(), evs); err != nil {
+		t.Fatalf("handle: %v", err)
+	}
+	if s.count != 0 { // dry-run should skip sends
+		t.Fatalf("expected no sends in dry-run, got %d", s.count)
+	}
+
+	// now run non-dry and ensure dedupe prevents duplicate
+	runner.dryRun = false
+	if err := runner.handleEvents(context.Background(), evs); err != nil {
+		t.Fatalf("handle: %v", err)
+	}
+	if s.count != 1 {
+		t.Fatalf("expected 1 send, got %d", s.count)
+	}
+	if err := runner.handleEvents(context.Background(), evs); err != nil {
+		t.Fatalf("handle dup: %v", err)
+	}
+	if s.count != 1 {
+		t.Fatalf("expected dedupe to skip duplicate send")
+	}
+}
+
+func newTestStore(t *testing.T) *storage.Store {
+	t.Helper()
+	store, err := storage.Open(t.TempDir() + "/db.sqlite")
+	if err != nil {
+		t.Fatalf("open store: %v", err)
+	}
+	t.Cleanup(func() { _ = store.Close() })
+	return store
+}

internal/health/server.go

@@ -0,0 +1,55 @@
+package health
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"time"
+)
+
+type Checker struct {
+	DBPing  func(ctx context.Context) error
+	RPCPing func(ctx context.Context) error
+}
+
+// Serve starts a minimal /healthz handler.
+func Serve(addr string, checker Checker) *http.Server {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
+		defer cancel()
+
+		status := map[string]string{"status": "ok"}
+		code := http.StatusOK
+
+		if checker.DBPing != nil {
+			if err := checker.DBPing(ctx); err != nil {
+				status["db"] = "fail"
+				code = http.StatusServiceUnavailable
+			} else {
+				status["db"] = "ok"
+			}
+		}
+		if checker.RPCPing != nil {
+			if err := checker.RPCPing(ctx); err != nil {
+				status["rpc"] = "fail"
+				code = http.StatusServiceUnavailable
+			} else {
+				status["rpc"] = "ok"
+			}
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(code)
+		_ = json.NewEncoder(w).Encode(status)
+	})
+
+	srv := &http.Server{
+		Addr:              addr,
+		Handler:           mux,
+		ReadHeaderTimeout: 3 * time.Second,
+	}
+	go func() { _ = srv.ListenAndServe() }()
+	return srv
+}
+

internal/logging/logging.go

@@ -0,0 +1,27 @@
+package logging
+
+import (
+	"log/slog"
+	"os"
+	"strings"
+)
+
+// New returns a minimal structured logger with secret redaction.
+func New() *slog.Logger {
+	handler := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelInfo,
+		ReplaceAttr: func(groups []string, a slog.Attr) slog.Attr {
+			if isSecretKey(a.Key) {
+				a.Value = slog.StringValue("[redacted]")
+			}
+			return a
+		},
+	})
+	return slog.New(handler)
+}
+
+func isSecretKey(k string) bool {
+	k = strings.ToLower(k)
+	return strings.Contains(k, "token") || strings.Contains(k, "secret") || strings.Contains(k, "key") || strings.Contains(k, "pass")
+}
+

tasks.md

@@ -115,8 +115,8 @@ sinks:
   - [x] Templates via `text/template`; helpers `pretty_json`, `short_addr`.
   - AC: Duplicate events do not resend (dedupe key honored); sink errors retried with backoff.
 - Phase H — Replay & Dry-Run
-  - [ ] `--from/--to` historical scan; progress output; backpressure.
-  - [ ] `--dry-run` writes alerts only; no network sends.
+  - [x] `--from/--to` historical scan; progress output; backpressure.
+  - [x] `--dry-run` writes alerts only; no network sends.
   - AC: 5k-block replay yields expected count; zero sink calls in dry-run; deterministic reruns.
 - Phase I — Ops (only needed)
   - [ ] `/healthz` HTTP (db, RPCs); optional `/metrics` Prometheus counters.