Signed-off-by: kaedwen <kaedwen@heinrich.blue>
This commit is contained in:
4 files changed
+45
-9
No files matched your search
+1
-1
@@ -74,7 +74,7 @@ func New(cfg *config.Config, logger *slog.Logger) (*Application, error) {
|
|||||||
if fetchTimeout == 0 {
|
if fetchTimeout == 0 {
|
||||||
fetchTimeout = 5 * time.Minute // Default timeout
|
fetchTimeout = 5 * time.Minute // Default timeout
|
||||||
}
|
}
|
||||||
newsAgg := news.NewAggregator(newsRepo, cfg.News.PollInterval.Duration, fetchTimeout, llmScorer, logger)
|
newsAgg := news.NewAggregator(newsRepo, cfg.News.PollInterval.Duration, fetchTimeout, cfg.News.RetentionDays, llmScorer, logger)
|
||||||
|
|
||||||
// Add news sources from config
|
// Add news sources from config
|
||||||
if len(cfg.News.Sources) > 0 {
|
if len(cfg.News.Sources) > 0 {
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ type Aggregator struct {
|
|||||||
llmScorer *LLMScorer
|
llmScorer *LLMScorer
|
||||||
pollInterval time.Duration
|
pollInterval time.Duration
|
||||||
fetchTimeout time.Duration
|
fetchTimeout time.Duration
|
||||||
|
retentionDays int
|
||||||
logger *slog.Logger
|
logger *slog.Logger
|
||||||
onNewsUpdated func() // Callback for SSE notifications
|
onNewsUpdated func() // Callback for SSE notifications
|
||||||
}
|
}
|
||||||
@@ -27,17 +28,19 @@ func NewAggregator(
|
|||||||
newsRepo *db.NewsRepository,
|
newsRepo *db.NewsRepository,
|
||||||
pollInterval time.Duration,
|
pollInterval time.Duration,
|
||||||
fetchTimeout time.Duration,
|
fetchTimeout time.Duration,
|
||||||
|
retentionDays int,
|
||||||
llmScorer *LLMScorer,
|
llmScorer *LLMScorer,
|
||||||
logger *slog.Logger,
|
logger *slog.Logger,
|
||||||
) *Aggregator {
|
) *Aggregator {
|
||||||
agg := &Aggregator{
|
agg := &Aggregator{
|
||||||
sources: make([]Source, 0),
|
sources: make([]Source, 0),
|
||||||
newsRepo: newsRepo,
|
newsRepo: newsRepo,
|
||||||
analyzer: NewAnalyzer(),
|
analyzer: NewAnalyzer(),
|
||||||
llmScorer: llmScorer,
|
llmScorer: llmScorer,
|
||||||
pollInterval: pollInterval,
|
pollInterval: pollInterval,
|
||||||
fetchTimeout: fetchTimeout,
|
fetchTimeout: fetchTimeout,
|
||||||
logger: logger,
|
retentionDays: retentionDays,
|
||||||
|
logger: logger,
|
||||||
}
|
}
|
||||||
|
|
||||||
return agg
|
return agg
|
||||||
@@ -222,6 +225,20 @@ func (a *Aggregator) fetchAllSources(ctx context.Context) error {
|
|||||||
a.onNewsUpdated()
|
a.onNewsUpdated()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Clean up old news articles if retention is configured
|
||||||
|
if a.retentionDays > 0 {
|
||||||
|
cutoff := time.Now().AddDate(0, 0, -a.retentionDays)
|
||||||
|
deleted, err := a.newsRepo.DeleteOlderThan(ctx, cutoff)
|
||||||
|
if err != nil {
|
||||||
|
a.logger.Error("failed to cleanup old news", slog.Any("error", err))
|
||||||
|
} else if deleted > 0 {
|
||||||
|
a.logger.Info("cleaned up old news",
|
||||||
|
slog.Int64("deleted", deleted),
|
||||||
|
slog.Int("retention_days", a.retentionDays),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -67,6 +67,7 @@ type OIDCConfig struct {
|
|||||||
type NewsConfig struct {
|
type NewsConfig struct {
|
||||||
PollInterval Duration `yaml:"poll_interval"`
|
PollInterval Duration `yaml:"poll_interval"`
|
||||||
FetchTimeout Duration `yaml:"fetch_timeout"` // Timeout per source (fetch + processing)
|
FetchTimeout Duration `yaml:"fetch_timeout"` // Timeout per source (fetch + processing)
|
||||||
|
RetentionDays int `yaml:"retention_days"` // Drop news older than this many days
|
||||||
DefaultRateLimit *RateLimit `yaml:"default_rate_limit,omitempty"`
|
DefaultRateLimit *RateLimit `yaml:"default_rate_limit,omitempty"`
|
||||||
Sources []NewsSource `yaml:"sources"`
|
Sources []NewsSource `yaml:"sources"`
|
||||||
}
|
}
|
||||||
@@ -151,7 +152,8 @@ func Load() (*Config, error) {
|
|||||||
PollInterval: Duration{
|
PollInterval: Duration{
|
||||||
Duration: time.Duration(getEnvInt("NEWS_POLL_INTERVAL", 300)) * time.Second,
|
Duration: time.Duration(getEnvInt("NEWS_POLL_INTERVAL", 300)) * time.Second,
|
||||||
},
|
},
|
||||||
Sources: []NewsSource{}, // Will be loaded from YAML
|
RetentionDays: getEnvInt("NEWS_RETENTION_DAYS", 7),
|
||||||
|
Sources: []NewsSource{}, // Will be loaded from YAML
|
||||||
},
|
},
|
||||||
LLMScorer: LLMScorerConfig{
|
LLMScorer: LLMScorerConfig{
|
||||||
Enabled: getEnvBool("LLM_SCORER_ENABLED", false),
|
Enabled: getEnvBool("LLM_SCORER_ENABLED", false),
|
||||||
|
|||||||
@@ -171,3 +171,20 @@ func (r *NewsRepository) CountRecentByTime(ctx context.Context, since time.Time)
|
|||||||
|
|
||||||
return count, nil
|
return count, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// DeleteOlderThan deletes news articles published before the given time
|
||||||
|
func (r *NewsRepository) DeleteOlderThan(ctx context.Context, before time.Time) (int64, error) {
|
||||||
|
query := `DELETE FROM news_articles WHERE published_at < ?`
|
||||||
|
|
||||||
|
result, err := r.db.ExecContext(ctx, query, before)
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("failed to delete old news: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
rowsAffected, err := result.RowsAffected()
|
||||||
|
if err != nil {
|
||||||
|
return 0, fmt.Errorf("failed to get rows affected: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return rowsAffected, nil
|
||||||
|
}
|
||||||
Reference in new issue
Block a user