✨ Introduce seed-telegrams tool for generating synthetic telegrams for testing and development. Enhance README with detailed usage instructions, command-line parameters, and examples. Add unit tests for telegram generation and status selection logic to ensure robustness. Update documentation to reflect new features and usage scenarios.

2025-11-16 17:24:16 +08:00
parent 06b2495d77
commit 12fda00df9
9 changed files with 730 additions and 80 deletions
@@ -532,6 +532,87 @@ Flight plan messages contain detailed flight planning information.
 -<OtherInfo>)
 ```

+## Seed Tool: `cmd/seed-telegrams`
+
+`seed-telegrams` 是一个开发/测试用的报文发生器，用来向 NATS/JetStream 持续或突发地发送合成电报（ARR/DEP/CNL/DLA/FPL），用于驱动解析与下游流水线。
+
+### 支持的类别与内容
+
+生成的电报遵循与解析器相同的格式约定：
+
+- ARR / DEP：支持无 SSR、合法简单 SSR，以及刻意构造为“当前正则无法解析”的复杂 SSR。
+- CNL / DLA：与 `internal/parsers/aviation_parser_test.go` 中的测试样例同一类结构。
+- FPL：生成包含多行 route 与 `OtherInfo` 字段的完整 FPL，`OtherInfo` 中会随机组合 `PBN/`, `NAV/`, `REG/`, `EET/`, `SEL/`, `PER/`, `RIF/`, `RMK/` 等片段，以覆盖解析逻辑。
+
+### 命令行参数
+
+常用参数：
+
+- `--nats-url`：NATS 地址（默认 `nats://127.0.0.1:4222`，为空字符串则完全不连接 NATS）。
+- `--subject`：普通 NATS 发布 subject（默认 `telegram.raw`）。
+- `--jetstream`：是否使用 JetStream 发布。
+- `--stream` / `--js-subject`：JetStream 相关选项。
+- `--count`：要发送的电报数量（`mode=burst` 或有上限的 interval/mixed 时生效）。
+- `--category`：`ARR|DEP|CNL|DLA|FPL|mixed`，`mixed` 表示在五类中随机选择。
+- `--status`：`parsed|header_error|body_error|publish_error|repository_error|random`。
+  - `random` 模式下，合法报文偏向标记为 `parsed`，刻意非法报文偏向标记为 `body_error`。
+- `--error-reason`：错误原因说明，将写入元数据 header（默认 `synthetic test payload`）。
+- `--dry-run`：只打印电报内容，不真正发布到 NATS。
+- `--header-format`：`json|none`，控制是否以 JSON 形式附加元数据 header。
+- `--mode`：seed 模式：
+  - `burst`：一次性快速发送完 `count` 条。
+  - `interval`：按照给定时间间隔持续发送。
+  - `mixed`：先按 interval 发送一部分，再以 burst 方式发送剩余。
+- `--interval-min` / `--interval-max`：`interval`/`mixed` 模式下两条电报之间的最小/最大间隔（默认 `1s` / `2s`）。
+- `--duration`：`interval`/`mixed` 模式下的总持续时间，`0` 表示仅按 `count` 控制停止条件。
+
+### 使用示例
+
+#### 1. 一次性快速打 100 条（突发流量）
+
+```bash
+go run ./cmd/seed-telegrams \
+  --count=100 \
+  --mode=burst \
+  --category=mixed \
+  --status=random
+```
+
+#### 2. 模拟真实流量：每 1–2 秒发一条，持续 5 分钟
+
+```bash
+go run ./cmd/seed-telegrams \
+  --mode=interval \
+  --interval-min=1s \
+  --interval-max=2s \
+  --duration=5m \
+  --status=random \
+  --category=mixed
+```
+
+#### 3. 慢热 + 突发：前半段慢慢发，后半段瞬间打完
+
+```bash
+go run ./cmd/seed-telegrams \
+  --count=200 \
+  --mode=mixed \
+  --interval-min=500ms \
+  --interval-max=1500ms \
+  --status=random
+```
+
+#### 4. 只打印合成电报，不发送（本地调试报文格式）
+
+```bash
+go run ./cmd/seed-telegrams \
+  --count=5 \
+  --mode=burst \
+  --dry-run \
+  --category=FPL
+```
+
+该工具专门为开发与测试设计，不影响生产服务逻辑，推荐在本地或测试环境配合解析与存储流水线一起使用，用于回归测试、吞吐量观察和错误场景演练。
+
 **Parsed Fields:**
 - `category`: "FPL"
 - `flight_number`: Flight number
@@ -23,6 +23,26 @@ var (
 	bodyCategories     = []string{"ARR", "DEP", "CNL", "DLA", "FPL"}
 )

+// SeedConfig controls how telegrams are generated and dispatched.
+type SeedConfig struct {
+	Count        int
+	Mode         string
+	IntervalMin  time.Duration
+	IntervalMax  time.Duration
+	Duration     time.Duration
+	CategoryFlag string
+	StatusFlag   string
+	ErrorReason  string
+	HeaderFormat string
+	DryRun       bool
+}
+
+// publishFunc abstracts the publish side-effect so that it can be swapped in tests.
+type publishFunc func(*telegram) error
+
+// sleepFunc is used instead of time.Sleep so tests can stub out real sleeping.
+var sleepFunc = time.Sleep
+
 func main() {
 	natsURL := flag.String("nats-url", "nats://127.0.0.1:4222", "NATS server URL (empty skips publish)")
 	subject := flag.String("subject", "telegram.raw", "Subject to publish telegrams to")
@@ -36,6 +56,10 @@ func main() {
 	jsStream := flag.String("stream", "", "JetStream stream (optional when --jetstream)")
 	jsSubject := flag.String("js-subject", "", "Override subject for JetStream publish (defaults to --subject)")
 	headerFormat := flag.String("header-format", "json", "Metadata header encoding: json|none")
+	mode := flag.String("mode", "burst", "Seed mode: burst|interval|mixed")
+	intervalMin := flag.Duration("interval-min", time.Second, "Minimum interval between messages in interval/mixed modes")
+	intervalMax := flag.Duration("interval-max", 2*time.Second, "Maximum interval between messages in interval/mixed modes")
+	duration := flag.Duration("duration", 0, "Total duration for interval/mixed modes (0 = rely on --count only)")
 	flag.Parse()

 	rand.Seed(time.Now().UnixNano())
@@ -64,62 +88,34 @@ func main() {
 		}
 	}

-	categories := bodyCategories
-	if strings.ToLower(*category) != "mixed" {
-		categories = []string{strings.ToUpper(*category)}
+	// Normalize and prepare configuration.
+	if *intervalMax < *intervalMin {
+		*intervalMax = *intervalMin
 	}

-	statuses := statusValues
-	if strings.ToLower(*status) != "random" {
-		statuses = []string{strings.ToLower(*status)}
+	cfg := SeedConfig{
+		Count:        *count,
+		Mode:         strings.ToLower(*mode),
+		IntervalMin:  *intervalMin,
+		IntervalMax:  *intervalMax,
+		Duration:     *duration,
+		CategoryFlag: *category,
+		StatusFlag:   *status,
+		ErrorReason:  *errorReason,
+		HeaderFormat: *headerFormat,
+		DryRun:       *dryRun,
 	}

-	for i := 0; i < *count; i++ {
-		cat := categories[rand.Intn(len(categories))]
-		payload, intentionallyInvalid := buildTelegram(cat)
+	categories := buildCategories(cfg.CategoryFlag, bodyCategories)
+	statuses := buildStatuses(cfg.StatusFlag, statusValues)

-		// 当状态为 random 时，根据报文是否合法来倾向选择 parsed 或 body_error
-		if strings.ToLower(*status) == "random" {
-			if intentionallyInvalid {
-				// 故意非法的报文：大概率标记为 body_error
-				if rand.Intn(100) < 80 {
-					payload.Status = "body_error"
-				} else {
-					payload.Status = statusValues[rand.Intn(len(statusValues))]
-				}
-			} else {
-				// 合法报文：大概率标记为 parsed
-				if rand.Intn(100) < 70 {
-					payload.Status = "parsed"
-				} else {
-					payload.Status = statusValues[rand.Intn(len(statusValues))]
-				}
-			}
-		} else {
-			// 非 random 模式下沿用原有逻辑
-			payload.Status = statuses[rand.Intn(len(statuses))]
-		}
-
-		payload.ErrorReason = *errorReason
-		payload.Metadata = map[string]string{
-			"message_id": payload.MessageID,
-			"category":   payload.Category,
-			"comments":   fmt.Sprintf("seeded iteration=%d", i),
-			"status":     payload.Status,
-		}
-
-		if *dryRun {
-			blob, _ := json.MarshalIndent(payload, "", "  ")
-			fmt.Println(string(blob))
-			fmt.Println("---")
-			continue
-		}
-
-		if nc != nil && !*noNATS {
+	var publisher publishFunc
+	if nc != nil && !*noNATS {
+		publisher = func(payload *telegram) error {
 			data := []byte(payload.Content)
 			msg := &nats.Msg{Subject: *subject, Data: data, Header: nats.Header{}}
 			msg.Header.Set("Nats-Msg-Id", payload.UUID)
-			if strings.ToLower(*headerFormat) == "json" {
+			if strings.ToLower(cfg.HeaderFormat) == "json" {
 				headerJSON, _ := json.Marshal(payload.Metadata)
 				msg.Header.Set("x-telegram-meta", string(headerJSON))
 			}
@@ -134,16 +130,22 @@ func main() {
 				}
 				msg.Subject = pubSubject
 				if _, err := js.PublishMsg(msg); err != nil {
-					log.Fatalf("jetstream publish: %v", err)
-				}
-			} else {
-				if err := nc.PublishMsg(msg); err != nil {
-					log.Fatalf("nats publish: %v", err)
+					return fmt.Errorf("jetstream publish: %w", err)
 				}
+				return nil
 			}
+
+			if err := nc.PublishMsg(msg); err != nil {
+				return fmt.Errorf("nats publish: %w", err)
+			}
+			return nil
 		}
 	}

+	if err := RunSeed(cfg, categories, statuses, publisher); err != nil {
+		log.Fatalf("run seed: %v", err)
+	}
+
 	if !*dryRun {
 		if nc != nil && !*noNATS {
 			log.Printf("Published %d telegram(s) to %s", *count, *subject)
@@ -151,6 +153,160 @@ func main() {
 	}
 }

+// RunSeed drives telegram generation according to the configured mode.
+// It delegates message construction to buildTelegram and side-effects to the provided publisher.
+func RunSeed(cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error {
+	if len(categories) == 0 {
+		return fmt.Errorf("no categories available")
+	}
+	if len(statuses) == 0 {
+		return fmt.Errorf("no statuses available")
+	}
+
+	switch cfg.Mode {
+	case "interval":
+		return runIntervalMode(cfg, categories, statuses, publisher)
+	case "mixed":
+		return runMixedMode(cfg, categories, statuses, publisher)
+	default:
+		// Default to burst semantics.
+		return runBurstMode(cfg, categories, statuses, publisher)
+	}
+}
+
+func runBurstMode(cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error {
+	count := cfg.Count
+	if count <= 0 {
+		return nil
+	}
+	for i := 0; i < count; i++ {
+		if err := sendTelegram(i, cfg, categories, statuses, publisher); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func runIntervalMode(cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error {
+	start := time.Now()
+	sent := 0
+
+	for {
+		if cfg.Count > 0 && sent >= cfg.Count {
+			break
+		}
+		if cfg.Duration > 0 && time.Since(start) >= cfg.Duration {
+			break
+		}
+
+		if err := sendTelegram(sent, cfg, categories, statuses, publisher); err != nil {
+			return err
+		}
+		sent++
+
+		// Compute next sleep duration within [IntervalMin, IntervalMax].
+		sleepDur := cfg.IntervalMin
+		if cfg.IntervalMax > cfg.IntervalMin {
+			delta := cfg.IntervalMax - cfg.IntervalMin
+			sleepDur = cfg.IntervalMin + time.Duration(rand.Int63n(int64(delta)+1))
+		}
+		if sleepDur > 0 && !cfg.DryRun {
+			sleepFunc(sleepDur)
+		}
+	}
+
+	return nil
+}
+
+// mixed 模式：前半段使用 interval 模式，后半段使用 burst。
+func runMixedMode(cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error {
+	// 简单策略：如果 Count>0，前半部分 interval，后半部分 burst；否则退化为 interval。
+	if cfg.Count <= 0 {
+		return runIntervalMode(cfg, categories, statuses, publisher)
+	}
+
+	half := cfg.Count / 2
+	if half == 0 {
+		// Count==1 时直接按 burst 处理。
+		return runBurstMode(cfg, categories, statuses, publisher)
+	}
+
+	intervalCfg := cfg
+	intervalCfg.Count = half
+	if err := runIntervalMode(intervalCfg, categories, statuses, publisher); err != nil {
+		return err
+	}
+
+	burstCfg := cfg
+	burstCfg.Count = cfg.Count - half
+	return runBurstMode(burstCfg, categories, statuses, publisher)
+}
+
+// sendTelegram builds a single telegram, assigns status/metadata, and either prints or publishes it.
+func sendTelegram(iteration int, cfg SeedConfig, categories []string, statuses []string, publisher publishFunc) error {
+	cat := categories[rand.Intn(len(categories))]
+	payload, intentionallyInvalid := buildTelegram(cat)
+
+	payload.Status = chooseStatus(intentionallyInvalid, cfg.StatusFlag, statuses)
+	payload.ErrorReason = cfg.ErrorReason
+	payload.Metadata = map[string]string{
+		"message_id": payload.MessageID,
+		"category":   payload.Category,
+		"comments":   fmt.Sprintf("seeded iteration=%d", iteration),
+		"status":     payload.Status,
+	}
+
+	if cfg.DryRun {
+		blob, _ := json.MarshalIndent(payload, "", "  ")
+		fmt.Println(string(blob))
+		fmt.Println("---")
+		return nil
+	}
+
+	if publisher != nil {
+		return publisher(payload)
+	}
+	return nil
+}
+
+// buildCategories returns the effective categories list based on the CLI flag.
+func buildCategories(flagValue string, all []string) []string {
+	if strings.ToLower(flagValue) == "mixed" || flagValue == "" {
+		return all
+	}
+	return []string{strings.ToUpper(flagValue)}
+}
+
+// buildStatuses returns the effective statuses list based on the CLI flag.
+func buildStatuses(flagValue string, all []string) []string {
+	if strings.ToLower(flagValue) == "random" || flagValue == "" {
+		return all
+	}
+	return []string{strings.ToLower(flagValue)}
+}
+
+// chooseStatus encapsulates the status selection logic, including the special \"random\" behaviour.
+func chooseStatus(intentionallyInvalid bool, statusFlag string, statuses []string) string {
+	if strings.ToLower(statusFlag) != "random" {
+		return statuses[rand.Intn(len(statuses))]
+	}
+
+	// random 模式：根据报文是否合法，对 parsed/body_error 做倾向性选择。
+	if intentionallyInvalid {
+		// 非法报文：大概率 body_error。
+		if rand.Intn(100) < 80 {
+			return "body_error"
+		}
+	} else {
+		// 合法报文：大概率 parsed。
+		if rand.Intn(100) < 70 {
+			return "parsed"
+		}
+	}
+
+	return statuses[rand.Intn(len(statuses))]
+}
+
 type telegram struct {
 	UUID        string            `json:"uuid"`
 	MessageID   string            `json:"message_id"`
@@ -0,0 +1,211 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+// Test buildBody generates syntactically plausible bodies for each category.
+func TestBuildBodyFormats(t *testing.T) {
+	tests := []struct {
+		category string
+		prefix   string
+		suffix   string
+	}{
+		{"ARR", "(ARR-", ")"},
+		{"DEP", "(DEP-", ")"},
+		{"CNL", "(CNL-", ")"},
+		{"DLA", "(DLA-", ")"},
+		{"FPL", "(FPL-", ")"},
+	}
+
+	for _, tt := range tests {
+		body, invalid := buildBody(tt.category)
+		if !invalid && body == "" {
+			t.Fatalf("category %s: expected non-empty body", tt.category)
+		}
+		if body[0:len(tt.prefix)] != tt.prefix {
+			t.Fatalf("category %s: expected prefix %q, got %q", tt.category, tt.prefix, body[0:len(tt.prefix)])
+		}
+		if body[len(body)-len(tt.suffix):] != tt.suffix {
+			t.Fatalf("category %s: expected suffix %q, got %q", tt.category, tt.suffix, body[len(body)-len(tt.suffix):])
+		}
+	}
+}
+
+// Test buildTelegram wires header and body into a full telegram.
+func TestBuildTelegramBasicFields(t *testing.T) {
+	tg, _ := buildTelegram("ARR")
+	if tg.Category != "ARR" {
+		t.Fatalf("expected category ARR, got %s", tg.Category)
+	}
+	if tg.MessageID == "" {
+		t.Fatalf("expected non-empty MessageID")
+	}
+	if tg.UUID == "" {
+		t.Fatalf("expected non-empty UUID")
+	}
+	if tg.Content == "" {
+		t.Fatalf("expected non-empty Content")
+	}
+	if tg.ReceivedAt.IsZero() {
+		t.Fatalf("expected ReceivedAt to be set")
+	}
+	if tg.Metadata != nil {
+		t.Fatalf("expected Metadata to be nil from buildTelegram")
+	}
+}
+
+func TestBuildCategories(t *testing.T) {
+	all := []string{"ARR", "DEP", "CNL", "DLA", "FPL"}
+
+	got := buildCategories("mixed", all)
+	if len(got) != len(all) {
+		t.Fatalf("expected all categories for mixed, got %d", len(got))
+	}
+
+	got = buildCategories("", all)
+	if len(got) != len(all) {
+		t.Fatalf("expected all categories for empty flag, got %d", len(got))
+	}
+
+	got = buildCategories("arr", all)
+	if len(got) != 1 || got[0] != "ARR" {
+		t.Fatalf("expected single category ARR, got %#v", got)
+	}
+}
+
+func TestBuildStatuses(t *testing.T) {
+	all := []string{"parsed", "header_error", "body_error"}
+
+	got := buildStatuses("random", all)
+	if len(got) != len(all) {
+		t.Fatalf("expected all statuses for random, got %d", len(got))
+	}
+
+	got = buildStatuses("", all)
+	if len(got) != len(all) {
+		t.Fatalf("expected all statuses for empty flag, got %d", len(got))
+	}
+
+	got = buildStatuses("parsed", all)
+	if len(got) != 1 || got[0] != "parsed" {
+		t.Fatalf("expected single status parsed, got %#v", got)
+	}
+}
+
+func TestChooseStatusNonRandomUsesProvidedList(t *testing.T) {
+	statuses := []string{"parsed"}
+	for i := 0; i < 10; i++ {
+		got := chooseStatus(false, "parsed", statuses)
+		if got != "parsed" {
+			t.Fatalf("expected parsed, got %s", got)
+		}
+	}
+}
+
+func TestChooseStatusRandomOnlyReturnsKnownStatuses(t *testing.T) {
+	statuses := []string{"parsed", "header_error", "body_error"}
+
+	for i := 0; i < 100; i++ {
+		got := chooseStatus(false, "random", statuses)
+		if !contains(statuses, got) && got != "body_error" && got != "parsed" {
+			t.Fatalf("unexpected status from chooseStatus: %s", got)
+		}
+	}
+}
+
+func contains(list []string, v string) bool {
+	for _, s := range list {
+		if s == v {
+			return true
+		}
+	}
+	return false
+}
+
+// Test RunSeed burst mode publishes Count messages.
+func TestRunSeedBurstPublishesCount(t *testing.T) {
+	cfg := SeedConfig{
+		Count:  5,
+		Mode:   "burst",
+		DryRun: false,
+	}
+	categories := []string{"ARR"}
+	statuses := []string{"parsed"}
+
+	var published int
+	pub := func(*telegram) error {
+		published++
+		return nil
+	}
+
+	if err := RunSeed(cfg, categories, statuses, pub); err != nil {
+		t.Fatalf("RunSeed burst returned error: %v", err)
+	}
+	if published != cfg.Count {
+		t.Fatalf("expected %d published messages, got %d", cfg.Count, published)
+	}
+}
+
+// Test RunSeed interval mode respects Count and does not sleep in DryRun.
+func TestRunSeedIntervalRespectsCount(t *testing.T) {
+	origSleep := sleepFunc
+	defer func() { sleepFunc = origSleep }()
+	sleepFunc = func(time.Duration) {}
+
+	cfg := SeedConfig{
+		Count:       3,
+		Mode:        "interval",
+		IntervalMin: 10 * time.Millisecond,
+		IntervalMax: 20 * time.Millisecond,
+		DryRun:      true,
+	}
+	categories := []string{"ARR"}
+	statuses := []string{"parsed"}
+
+	var published int
+	pub := func(*telegram) error {
+		published++
+		return nil
+	}
+
+	if err := RunSeed(cfg, categories, statuses, pub); err != nil {
+		t.Fatalf("RunSeed interval returned error: %v", err)
+	}
+	if published != cfg.Count {
+		t.Fatalf("expected %d published messages, got %d", cfg.Count, published)
+	}
+}
+
+// Test RunSeed mixed mode still results in exactly Count messages when Count>0.
+func TestRunSeedMixedPublishesCount(t *testing.T) {
+	origSleep := sleepFunc
+	defer func() { sleepFunc = origSleep }()
+	sleepFunc = func(time.Duration) {}
+
+	cfg := SeedConfig{
+		Count:       6,
+		Mode:        "mixed",
+		IntervalMin: 1 * time.Millisecond,
+		IntervalMax: 2 * time.Millisecond,
+		DryRun:      true,
+	}
+	categories := []string{"ARR"}
+	statuses := []string{"parsed"}
+
+	var published int
+	pub := func(*telegram) error {
+		published++
+		return nil
+	}
+
+	if err := RunSeed(cfg, categories, statuses, pub); err != nil {
+		t.Fatalf("RunSeed mixed returned error: %v", err)
+	}
+	if published != cfg.Count {
+		t.Fatalf("expected %d published messages, got %d", cfg.Count, published)
+	}
+}
+
+
@@ -31,6 +31,25 @@ Stop and clean the stack when finished:
 docker compose -f docker-compose.dev.yml down -v
 ```

+In development (`GO_ENV=dev` or unset), if you run the application while
+stopping and recreating the NATS/JetStream containers (for example:
+
+```bash
+docker compose -f docker-compose.dev.yml down -v
+docker compose -f docker-compose.dev.yml up -d postgres nats nats-box
+```
+
+), the JetStream state will be reset. The processor behaves as follows:
+
+- The NATS client keeps retrying the connection and automatically reconnects
+  when NATS is back.
+- The JetStream consumer detects missing streams/consumers and, in dev/test
+  environments, uses shared `EnsureStream`/`ensureConsumer` logic to
+  auto-recreate them.
+- In production environments, missing streams/consumers are treated as
+  configuration/operational errors and are not auto-recreated; operators
+  should investigate and fix the underlying issue.
+
 ### Using Taskfile shortcuts

 The `Taskfile.yml` includes helper targets that wrap the commands above:
@@ -101,6 +101,27 @@ Recommended pattern:
 - Use a backoff array such as `[5s, 30s, 2m]`.  
 - Treat messages that still fail after `max_deliver` as candidates for DLQ, via the permanent error/poison message path where applicable.

+### JetStream Availability and Auto-Recovery (Dev vs Prod)
+
+- When the JetStream API is temporarily unavailable (for example, NATS has just
+  restarted and returns `ErrNoResponders`), the consumer uses an exponential
+  backoff when retrying `Fetch` calls (roughly `1s, 2s, 4s, ...` up to
+  around `30s`) to avoid log spam while allowing the system to recover.
+- In dev/test environments, if the stream or consumer is detected as missing at
+  runtime (for example after `docker compose down -v`), the consumer calls the
+  shared `EnsureStream` and `ensureConsumer` logic to recreate them and
+  re-establish subscriptions.
+- In production environments, missing streams/consumers are treated as
+  configuration or operational errors:
+  - They are **not** auto-recreated.
+  - Errors are logged prominently so operators can diagnose and fix the issue.
+- On the publishing side, JetStream `ErrNoResponders` and similar errors are
+  treated as temporary by the processor:
+  - Such errors cause the consumer to NAK messages and rely on the configured
+    backoff for retries.
+  - Permanent configuration/permission errors remain mapped to permanent
+    failures and follow the DLQ + ACK flow.
+
 ### Alerts and Dashboards

 Prometheus alert suggestions:
@@ -169,7 +169,17 @@ func (p *MessageProcessor) Handle(ctx context.Context, raw []byte, msgID string)
 		p.telemetry.RecordFailure("publisher")
 		p.telemetry.RecordProcessingResult(ctx, string(parsed.Status), parsed.Category, latency)
 		p.persistRaw(ctx, parsed)
-		// Mark as permanent so the consumer will ack instead of retrying
+
+		// Treat clearly temporary JetStream issues (e.g. no responders) as transient so
+		// the consumer will NAK and retry according to backoff settings.
+		lowerErr := strings.ToLower(err.Error())
+		if strings.Contains(lowerErr, "no responders") {
+			pubSpan.End()
+			// Return a non-permanent error to trigger retry via nakWithStrategy in the consumer.
+			return fmt.Errorf("transient publish error: %w", err)
+		}
+
+		// Other publish errors are treated as permanent and will go to DLQ + ACK.
 		pubSpan.End()
 		return Permanent(fmt.Errorf("failed to publish message: %w", err))
 	}
@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"os"
 	"strings"
 	"time"

@@ -49,6 +50,32 @@ type Consumer struct {
 	consecutiveProcessErrors int
 }

+func isDevLikeEnv() bool {
+	switch strings.ToLower(os.Getenv("GO_ENV")) {
+	case "", "dev", "development", "test", "testing":
+		return true
+	default:
+		return false
+	}
+}
+
+func isJetStreamResourceNotFound(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, nats.ErrStreamNotFound) || errors.Is(err, nats.ErrConsumerNotFound) {
+		return true
+	}
+
+	// Some JetStream API errors are only exposed via error strings.
+	msg := strings.ToLower(err.Error())
+	if strings.Contains(msg, "stream not found") || strings.Contains(msg, "consumer not found") {
+		return true
+	}
+
+	return false
+}
+
 // ProvideConsumer creates a NATS consumer
 func ProvideConsumer(
 	conn *nats.Conn,
@@ -201,6 +228,58 @@ func (c *Consumer) ensureConsumer() error {
 	return nil
 }

+// recoverJetStreamResources attempts to recreate the stream and consumer in
+// dev/test environments if they are missing. It is safe to call multiple times.
+func (c *Consumer) recoverJetStreamResources() error {
+	if c.js == nil {
+		return fmt.Errorf("jetstream context is nil")
+	}
+	if c.cfg == nil {
+		return fmt.Errorf("config is nil")
+	}
+
+	// Ensure stream exists (dev/test may auto-create, prod will error).
+	if err := EnsureStream(c.js, c.cfg, c.logger); err != nil {
+		return fmt.Errorf("ensure stream %s: %w", c.streamName, err)
+	}
+
+	// Ensure durable consumer exists and is properly bound.
+	if err := c.ensureConsumer(); err != nil {
+		return fmt.Errorf("ensure consumer %s: %w", c.consumerName, err)
+	}
+
+	return nil
+}
+
+// createPullSubscriptionWithRecovery creates a pull subscription and, in
+// dev/test environments, attempts to self-heal missing stream/consumer
+// by recreating them once.
+func (c *Consumer) createPullSubscriptionWithRecovery() (*nats.Subscription, error) {
+	sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName))
+	if err == nil {
+		return sub, nil
+	}
+
+	if isJetStreamResourceNotFound(err) && isDevLikeEnv() && shouldBootstrapStream() {
+		c.logger.Warn("PullSubscribe failed due to missing JetStream resources; attempting to recreate",
+			zap.Error(err),
+			zap.String("stream", c.streamName),
+			zap.String("consumer", c.consumerName),
+		)
+		if recErr := c.recoverJetStreamResources(); recErr != nil {
+			return nil, fmt.Errorf("failed to recover JetStream resources: %w", recErr)
+		}
+		// Retry subscription after successful recovery.
+		sub, err = c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName))
+		if err != nil {
+			return nil, fmt.Errorf("failed to create pull subscription after recovery: %w", err)
+		}
+		return sub, nil
+	}
+
+	return nil, fmt.Errorf("failed to create pull subscription: %w", err)
+}
+
 // validateDLQ verifies whether DLQ routing should be enabled and, if so, whether
 // the configured DLQ subject is bound to a JetStream stream. If validation fails,
 // DLQ routing is disabled (by clearing c.dlqSubject) and a warning is logged,
@@ -269,10 +348,10 @@ func (c *Consumer) Start(ctx context.Context) error {
 }

 func (c *Consumer) startJetStream(ctx context.Context) error {
-	// Create pull subscription
-	sub, err := c.js.PullSubscribe(c.subject, c.consumerName, nats.Bind(c.streamName, c.consumerName))
+	// Create pull subscription (with simple self-healing in dev/test).
+	sub, err := c.createPullSubscriptionWithRecovery()
 	if err != nil {
-		return fmt.Errorf("failed to create pull subscription: %w", err)
+		return err
 	}
 	defer sub.Unsubscribe()

@@ -296,6 +375,8 @@ func (c *Consumer) startJetStream(ctx context.Context) error {
 	defer statsCancel()
 	go c.emitConsumerStats(statsCtx)

+	var fetchErrorStreak int
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -308,25 +389,80 @@ func (c *Consumer) startJetStream(ctx context.Context) error {
 		msgs, err := sub.Fetch(c.batchSize, nats.MaxWait(c.batchTimeout))
 		if err != nil {
 			if errors.Is(err, nats.ErrTimeout) {
-				// Timeout is expected when no messages are available
+				// Timeout is expected when no messages are available.
 				continue
 			}
+
+			// JetStream API is currently unavailable (e.g., NATS just restarted or JetStream not ready).
 			if errors.Is(err, nats.ErrNoResponders) {
-				// JetStream API is currently unavailable (e.g., NATS just restarted or JetStream not ready).
-				// Back off a bit to avoid log spam while allowing the system to recover.
-				c.logger.Warn("JetStream not available, will retry",
+				fetchErrorStreak++
+				backoff := time.Duration(fetchErrorStreak) * time.Second
+				if backoff > 30*time.Second {
+					backoff = 30 * time.Second
+				}
+				c.logger.Warn("JetStream not available, will retry with backoff",
+					zap.Error(err),
+					zap.String("stream", c.streamName),
+					zap.String("consumer", c.consumerName),
+					zap.Duration("backoff", backoff),
+				)
+				time.Sleep(backoff)
+				continue
+			}
+
+			// Underlying consumer/stream removed while app is running.
+			if isJetStreamResourceNotFound(err) {
+				if isDevLikeEnv() && shouldBootstrapStream() {
+					c.logger.Warn("JetStream consumer or stream missing; attempting to recreate",
+						zap.Error(err),
+						zap.String("stream", c.streamName),
+						zap.String("consumer", c.consumerName),
+					)
+					if recErr := c.recoverJetStreamResources(); recErr != nil {
+						c.logger.Error("Failed to recover JetStream resources", zap.Error(recErr))
+						return recErr
+					}
+
+					// Recreate subscription after successful recovery.
+					sub.Unsubscribe()
+					sub, err = c.createPullSubscriptionWithRecovery()
+					if err != nil {
+						return err
+					}
+
+					// Reset error streak after successful recovery.
+					fetchErrorStreak = 0
+					continue
+				}
+
+				// Production: treat as configuration/operational error.
+				c.logger.Error("JetStream consumer or stream missing; not auto-recreating in this environment",
 					zap.Error(err),
 					zap.String("stream", c.streamName),
 					zap.String("consumer", c.consumerName),
 				)
-				time.Sleep(5 * time.Second)
-				continue
+				return err
 			}
-			c.logger.Error("Failed to fetch messages", zap.Error(err))
-			time.Sleep(time.Second)
+
+			// Generic error path with modest backoff.
+			fetchErrorStreak++
+			backoff := time.Duration(fetchErrorStreak) * time.Second
+			if backoff > 10*time.Second {
+				backoff = 10 * time.Second
+			}
+			c.logger.Error("Failed to fetch messages; backing off",
+				zap.Error(err),
+				zap.Duration("backoff", backoff),
+			)
+			time.Sleep(backoff)
 			continue
 		}

+		// Successful fetch -> reset error streak.
+		if fetchErrorStreak > 0 {
+			fetchErrorStreak = 0
+		}
+
 		// Process each message
 		// TODO: consider buffering messages to take advantage of Repository.InsertBatch for higher throughput.
 		for _, msg := range msgs {
@@ -45,15 +45,29 @@ func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (na
 		return nil, fmt.Errorf("failed to get JetStream context: %w", err)
 	}

-	// Create stream if it doesn't exist
+	// Ensure the stream exists and is minimally aligned with configuration.
+	if err := EnsureStream(js, cfg, logger); err != nil {
+		nc.Close()
+		return nil, err
+	}
+
+	return js, nil
+}
+
+// EnsureStream ensures that the configured JetStream stream exists and has
+// at least the expected subjects bound. It is safe to call multiple times.
+//
+// In dev/test environments (see shouldBootstrapStream), the stream will be
+// auto-created if it does not exist. In production, a missing stream results
+// in an error so that operators can intervene.
+func EnsureStream(js nats.JetStreamContext, cfg *config.Config, logger *zap.Logger) error {
 	streamName := cfg.NATS.Stream
 	consumerSubject := cfg.EffectiveSubscriptionTopic()
 	publisherSubject := strings.TrimSpace(cfg.Publisher.Topic)

 	streamSubjects := dedupeSubjects([]string{consumerSubject, publisherSubject})
 	if len(streamSubjects) == 0 {
-		nc.Close()
-		return nil, fmt.Errorf("no subjects configured for JetStream stream %s", streamName)
+		return fmt.Errorf("no subjects configured for JetStream stream %s", streamName)
 	}

 	streamLimits := cfg.NATS.StreamLimits
@@ -87,26 +101,22 @@ func ProvideJetStream(nc *nats.Conn, cfg *config.Config, logger *zap.Logger) (na
 		if errors.Is(err, nats.ErrStreamNotFound) {
 			if shouldBootstrapStream() {
 				if _, err = js.AddStream(streamConfig); err != nil {
-					nc.Close()
-					return nil, fmt.Errorf("failed to create stream: %w", err)
+					return fmt.Errorf("failed to create stream %s: %w", streamName, err)
 				}
-				logger.Info("Created JetStream",
+				logger.Info("Created JetStream stream",
 					zap.String("stream", streamName),
 					zap.Strings("subjects", streamSubjects),
 				)
-			} else {
-				nc.Close()
-				return nil, fmt.Errorf("stream %s not found and auto-creation disabled", streamName)
+				return nil
 			}
-		} else {
-			nc.Close()
-			return nil, fmt.Errorf("failed to fetch stream info: %w", err)
+			return fmt.Errorf("stream %s not found and auto-creation disabled", streamName)
 		}
-	} else {
-		validateStreamConfig(info, streamSubjects, logger)
+		return fmt.Errorf("failed to fetch stream info for %s: %w", streamName, err)
 	}

-	return js, nil
+	// Stream exists: validate subjects but do not fail hard if they differ.
+	validateStreamConfig(info, streamSubjects, logger)
+	return nil
 }

 func shouldBootstrapStream() bool {
@@ -5,7 +5,9 @@ import (
 	"caatsm/internal/infra/config"
 	"caatsm/internal/model"
 	"encoding/json"
+	"errors"
 	"fmt"
+
 	"github.com/google/uuid"
 	"github.com/nats-io/nats.go"
 	"go.uber.org/zap"
@@ -62,6 +64,10 @@ func (p *Publisher) Publish(message interface{}) error {
 	// Publish to JetStream
 	_, err = p.js.PublishMsg(jsMsg)
 	if err != nil {
+		// Distinguish temporary JetStream unavailability from permanent config errors.
+		if errors.Is(err, nats.ErrNoResponders) {
+			return fmt.Errorf("transient publish error (no responders): %w", err)
+		}
 		return fmt.Errorf("failed to publish message: %w", err)
 	}