package main import ( "encoding/json" "fmt" "os" "os/exec" "sort" "strings" "time" ) // broadcaster runs on startup and frequently checks if a broadcast is due func broadcaster(shutdown chan struct{}, dryRun bool) { logInfo("broadcaster", "Starting broadcaster", map[string]interface{}{ "startupDelay": int(STARTUP_DELAY.Seconds()), "checkInterval": BROADCAST_CHECK_INTERVAL.String(), "broadcastWindow": BROADCAST_INTERVAL.String(), "dryRun": dryRun, }) // Sleep on startup logInfo("broadcaster", "Sleeping before first broadcast check", map[string]interface{}{ "seconds": int(STARTUP_DELAY.Seconds()), }) select { case <-time.After(STARTUP_DELAY): // Continue after sleep case <-shutdown: logInfo("broadcaster", "Shutdown signal received during startup sleep", nil) return } // Initialize the last broadcast time from the database lastBroadcastTime := getLastBroadcastTime() if !lastBroadcastTime.IsZero() { logInfo("broadcaster", "Initialized last broadcast time from database", map[string]interface{}{ "lastBroadcastTime": lastBroadcastTime.Format(time.RFC3339), "timeSince": time.Since(lastBroadcastTime).String(), }) } else { logInfo("broadcaster", "No previous broadcast time found in database", nil) } // Run checks frequently ticker := time.NewTicker(BROADCAST_CHECK_INTERVAL) defer ticker.Stop() for { select { case <-ticker.C: now := time.Now() timeSinceLastBroadcast := now.Sub(lastBroadcastTime) // If it's been at least BROADCAST_INTERVAL since last broadcast // or if we haven't broadcast yet (lastBroadcastTime is zero) if lastBroadcastTime.IsZero() || timeSinceLastBroadcast >= BROADCAST_INTERVAL { logInfo("broadcaster", "Broadcast window reached, checking conditions", map[string]interface{}{ "timeSinceLastBroadcast": timeSinceLastBroadcast.String(), "requiredInterval": BROADCAST_INTERVAL.String(), }) // Only update lastBroadcastTime if we actually broadcast something if didBroadcast := checkAndBroadcast(dryRun); didBroadcast { lastBroadcastTime = now } } case <-shutdown: logInfo("broadcaster", "Shutting down broadcaster", nil) return } } } // checkAndBroadcast checks if there are any unsummarized articles before broadcasting // Returns true if a broadcast was made func checkAndBroadcast(dryRun bool) bool { // Check if there are any unsummarized articles articles := loadArticles() unsummarizedCount := 0 for _, article := range articles { if article.Summary == "" || article.Importance == 0 { unsummarizedCount++ } } if unsummarizedCount > 0 { logInfo("broadcaster", "Postponing broadcast - waiting for articles to be summarized", map[string]interface{}{ "unsummarizedCount": unsummarizedCount, "totalArticles": len(articles), }) return false } // No unsummarized articles, proceed with broadcast return broadcastWithRedundancyCheck(dryRun) } // broadcastWithRedundancyCheck attempts to find a non-redundant article to broadcast // Returns true if a broadcast was made func broadcastWithRedundancyCheck(dryRun bool) bool { articles := loadArticles() now := time.Now() cutoff := now.Add(-ARTICLE_FRESHNESS_WINDOW) // Time window for considering articles fresh var candidates []Article for _, a := range articles { // Only include articles that haven't been broadcast, have summaries, and are fresh if a.Summary != "" && a.BroadcastTime.IsZero() && a.FirstSeen.After(cutoff) { candidates = append(candidates, a) } } if len(candidates) == 0 { logInfo("broadcaster", "No fresh articles found for broadcasting", map[string]interface{}{ "totalArticles": len(articles), }) return false } // Sort by importance sort.Slice(candidates, func(i, j int) bool { return candidates[i].Importance > candidates[j].Importance }) // Get recent broadcasts recentBroadcasts := getRecentBroadcasts(12) // Try candidates until we find a non-redundant one for i := 0; i < len(candidates); i++ { candidate := candidates[i] // Check if this candidate would be redundant isRedundant, reason, err := checkRedundancy(candidate, recentBroadcasts) if err != nil { logInfo("redundancy", "Error checking redundancy, proceeding anyway", map[string]interface{}{ "error": err.Error(), "id": candidate.ID, }) // Continue with this candidate despite the error broadcastArticle(candidate, dryRun) return true } if isRedundant { logInfo("redundancy", "Article deemed redundant, marking as 'already broadcast'", map[string]interface{}{ "id": candidate.ID, "summary": candidate.Summary, "reason": reason, }) // Mark as broadcast with special timestamp to prevent future selection candidate.BroadcastTime = time.Unix(1, 0) // epoch + 1 second if err := updateArticle(candidate); err != nil { logInfo("redundancy", "Error updating article", map[string]interface{}{ "id": candidate.ID, "error": err.Error(), }) } continue // Try next candidate } // Not redundant, proceed with this candidate logInfo("redundancy", "Article passed redundancy check", map[string]interface{}{ "id": candidate.ID, "candidateNumber": i + 1, }) broadcastArticle(candidate, dryRun) return true } // If we got here, all candidates were redundant logInfo("broadcaster", "All candidates were deemed redundant, no broadcast", map[string]interface{}{ "candidatesChecked": len(candidates), }) return false } // broadcastArticle broadcasts the chosen article func broadcastArticle(chosen Article, dryRun bool) { // Get the abbreviated source name sourceAbbr := sourceAbbreviations[chosen.Source] if sourceAbbr == "" { // Default to first 3 characters if no abbreviation defined if len(chosen.Source) >= 3 { sourceAbbr = chosen.Source[:3] } else { sourceAbbr = chosen.Source } } ts := time.Now().Format("15:04 MST") msg := fmt.Sprintf("%s: [%s] [AI/LLM] %s", ts, sourceAbbr, strings.TrimSpace(chosen.Summary)) // Ensure message is under 200 characters if len(msg) > MAX_MESSAGE_LENGTH { // Calculate max summary length including timestamp, source prefix, AI/LLM prefix, and ellipsis maxSummaryLen := MAX_MESSAGE_LENGTH - len(ts) - len(sourceAbbr) - len(": [] [AI/LLM] ") - 3 // 3 for "..." truncatedSummary := strings.TrimSpace(chosen.Summary)[:maxSummaryLen] + "..." msg = fmt.Sprintf("%s: [%s] [AI/LLM] %s", ts, sourceAbbr, truncatedSummary) logInfo("broadcaster", "Message truncated to fit character limit", map[string]interface{}{ "limit": MAX_MESSAGE_LENGTH, "originalLength": len(chosen.Summary), "truncatedLength": maxSummaryLen, }) } // Print info about what will be broadcast cmdStr := fmt.Sprintf("meshtastic --sendtext \"%s\"", msg) logInfo("broadcaster", "Preparing to broadcast", map[string]interface{}{ "message": msg, "length": len(msg), "command": cmdStr, "dryRun": dryRun, "id": chosen.ID, "importance": chosen.Importance, "source": chosen.Source, "sourceAbbr": sourceAbbr, }) // Wait before broadcasting to allow time to see the message logInfo("broadcaster", "Waiting before broadcasting", map[string]interface{}{ "seconds": int(BROADCAST_PREPARATION_DELAY.Seconds()), }) time.Sleep(BROADCAST_PREPARATION_DELAY) // Update broadcast time and save to database chosen.BroadcastTime = time.Now() if err := updateArticle(chosen); err != nil { logInfo("broadcaster", "Error updating article broadcast time", map[string]interface{}{ "id": chosen.ID, "error": err.Error(), }) } logInfo("broadcaster", "Set broadcast time for article", map[string]interface{}{ "id": chosen.ID, "summary": chosen.Summary, "importance": chosen.Importance, "broadcastTime": chosen.BroadcastTime.Format(time.RFC3339), }) output, _ := json.MarshalIndent(chosen, "", " ") fmt.Println(string(output)) if !dryRun { logInfo("broadcaster", "Broadcasting message", map[string]interface{}{ "message": msg, "length": len(msg), }) cmd := exec.Command("meshtastic", "--sendtext", msg) cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr if err := cmd.Run(); err != nil { logInfo("broadcaster", "Broadcast failed", map[string]interface{}{ "error": err.Error(), "id": chosen.ID, }) } else { logInfo("broadcaster", "Broadcast success", map[string]interface{}{ "id": chosen.ID, }) } } else { // In dry-run mode, just print the command logInfo("broadcaster", "DRY RUN - would run command", map[string]interface{}{ "command": cmdStr, "id": chosen.ID, }) } }