nandi/glean · 0937d65

nandi/gleanpublic⑂ Fork 0

⑂ 0937d65

Keep articles read after a retention purge re-ingests them

read_state is keyed on the articles surrogate id. The retention purge
deleted old articles along with their read_state, but feeds keep serving
entries past the 30-day cutoff, so the next fetch re-inserted the same
entry under a fresh id with no read_state row. Every restart and cron
pass therefore resurrected the same backlog as unread, no matter how
recently the user had marked everything read.

Archive read state against the article's stable (feed_url, guid)
identity before the purge drops it, and restore it on ingest when a
purged entry comes back. The two touch points are the only writers, so
the history table cannot drift from read_state.

Also stop ingesting entries already past the retention window, so a
fetch no longer undoes the purge that just ran.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

nandi committed 2026-09-06T02:29:05-07:00 Browse files

0937d65 parent: 91d8ffd

modified internal/db/article.go +43 -0 Viewed

@@ -16,12 +16,22 @@ const articlesOrderByAsc = ` ORDER BY (CASE WHEN a.published > 'now' THEN 1 ELSE
16	16
17	17	type ArticleStore struct {
18	18	db *DB
	19	+ // retentionDays mirrors the article retention window. Ingest drops entries
	20	+ // already older than it, so the purge is not undone by the next fetch of a
	21	+ // feed that still serves its whole history. Zero disables the cutoff.
	22	+ retentionDays int
19	23	}
20	24
21	25	func NewArticleStore(db DB) ArticleStore {
22	26	return &ArticleStore{db: db}
23	27	}
24	28
	29	+// SetRetentionDays sets the ingest age cutoff. It must match the window passed
	30	+// to PurgeExpiredArticles.
	31	+func (s *ArticleStore) SetRetentionDays(days int) {
	32	+ s.retentionDays = days
	33	+}
	34	+
25	35	type Article struct {
26	36	ID int64
27	37	FeedURL string
@@ -61,6 +71,14 @@ func (s *ArticleStore) BatchUpsertArticles(ctx context.Context, articles []feed.
61	71	return nil
62	72	}
63	73
	74	+ // Anything already past the retention window would be deleted by the next
	75	+ // purge; ingesting it only churns the database and, before read state was
	76	+ // archived, resurfaced long-read articles as unread.
	77	+ var cutoff time.Time
	78	+ if s.retentionDays > 0 {
	79	+ cutoff = time.Now().AddDate(0, 0, -s.retentionDays)
	80	+ }
	81	+
64	82	tx, err := s.db.BeginTx(ctx, nil)
65	83	if err != nil {
66	84	return err
@@ -77,7 +95,12 @@ func (s *ArticleStore) BatchUpsertArticles(ctx context.Context, articles []feed.
77	95	}
78	96	defer stmt.Close()
79	97
	98	+ feedURLs := make(map[string]struct{}, 1)
80	99	for _, a := range articles {
	100	+ if !cutoff.IsZero() && !a.Published.IsZero() && a.Published.Before(cutoff) {
	101	+ continue
	102	+ }
	103	+ feedURLs[a.FeedURL] = struct{}{}
81	104	url := sql.NullString{String: a.URL, Valid: a.URL != ""}
82	105	author := sql.NullString{String: a.Author, Valid: a.Author != ""}
83	106	summary := sql.NullString{String: a.Summary, Valid: a.Summary != ""}
@@ -95,6 +118,26 @@ func (s *ArticleStore) BatchUpsertArticles(ctx context.Context, articles []feed.
95	118	}
96	119	}
97	120
	121	+ // Restore read state for entries that were purged and have now come back
	122	+ // under a new surrogate id. Without this they would read as unread again.
	123	+ restore, err := tx.PrepareContext(ctx, `
	124	+ INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
	125	+ SELECT h.user_did, a.id, h.is_read, h.read_at
	126	+ FROM articles.read_state_history h
	127	+ JOIN articles.articles a ON a.feed_url = h.feed_url AND a.guid = h.guid
	128	+ WHERE h.feed_url = ?
	129	+ ON CONFLICT(user_did, article_id) DO NOTHING
	130	+ `)
	131	+ if err != nil {
	132	+ return err
	133	+ }
	134	+ defer restore.Close()
	135	+ for feedURL := range feedURLs {
	136	+ if _, err := restore.ExecContext(ctx, feedURL); err != nil {
	137	+ return err
	138	+ }
	139	+ }
	140	+
98	141	return tx.Commit()
99	142	}
100	143

  type ArticleStore struct {
  	db *DB
++	// retentionDays mirrors the article retention window. Ingest drops entries
++	// already older than it, so the purge is not undone by the next fetch of a
++	// feed that still serves its whole history. Zero disables the cutoff.
++	retentionDays int
  }
  func NewArticleStore(db *DB) *ArticleStore {
  	return &ArticleStore{db: db}
  }
++// SetRetentionDays sets the ingest age cutoff. It must match the window passed
++// to PurgeExpiredArticles.
++func (s *ArticleStore) SetRetentionDays(days int) {
++	s.retentionDays = days
++}
++
  type Article struct {
  	ID             int64
  	FeedURL        string
  		return nil
  	}
++	// Anything already past the retention window would be deleted by the next
++	// purge; ingesting it only churns the database and, before read state was
++	// archived, resurfaced long-read articles as unread.
++	var cutoff time.Time
++	if s.retentionDays > 0 {
++		cutoff = time.Now().AddDate(0, 0, -s.retentionDays)
++	}
++
  	tx, err := s.db.BeginTx(ctx, nil)
  	if err != nil {
  		return err
  	}
  	defer stmt.Close()
++	feedURLs := make(map[string]struct{}, 1)
  	for _, a := range articles {
++		if !cutoff.IsZero() && !a.Published.IsZero() && a.Published.Before(cutoff) {
++			continue
++		}
++		feedURLs[a.FeedURL] = struct{}{}
  		url := sql.NullString{String: a.URL, Valid: a.URL != ""}
  		author := sql.NullString{String: a.Author, Valid: a.Author != ""}
  		summary := sql.NullString{String: a.Summary, Valid: a.Summary != ""}
  		}
  	}
++	// Restore read state for entries that were purged and have now come back
++	// under a new surrogate id. Without this they would read as unread again.
++	restore, err := tx.PrepareContext(ctx, `
++		INSERT INTO articles.read_state (user_did, article_id, is_read, read_at)
++		SELECT h.user_did, a.id, h.is_read, h.read_at
++		FROM articles.read_state_history h
++		JOIN articles.articles a ON a.feed_url = h.feed_url AND a.guid = h.guid
++		WHERE h.feed_url = ?
++		ON CONFLICT(user_did, article_id) DO NOTHING
++	`)
++	if err != nil {
++		return err
++	}
++	defer restore.Close()
++	for feedURL := range feedURLs {
++		if _, err := restore.ExecContext(ctx, feedURL); err != nil {
++			return err
++		}
++	}
++
  	return tx.Commit()
  }

modified internal/db/db.go +13 -0 Viewed

@@ -348,6 +348,19 @@ var articlesSchema = []string{
348	348	PRIMARY KEY (user_did, article_id)
349	349	)`,
350	350
	351	+ // read_state is keyed on the articles surrogate id, which does not survive
	352	+ // a retention purge and re-fetch of the same feed entry. read_state_history
	353	+ // records the same fact against the article's stable (feed_url, guid)
	354	+ // identity so "read" is not forgotten when a purged article is re-ingested.
	355	+ `CREATE TABLE IF NOT EXISTS articles.read_state_history (
	356	+ user_did TEXT NOT NULL,
	357	+ feed_url TEXT NOT NULL,
	358	+ guid TEXT NOT NULL,
	359	+ is_read BOOLEAN NOT NULL DEFAULT 0,
	360	+ read_at DATETIME,
	361	+ PRIMARY KEY (user_did, feed_url, guid)
	362	+ )`,
	363	+
351	364	`CREATE TABLE IF NOT EXISTS articles.annotations (
352	365	id INTEGER PRIMARY KEY AUTOINCREMENT,
353	366	uri TEXT NOT NULL UNIQUE,

  		PRIMARY KEY (user_did, article_id)
  	)`,
++	// read_state is keyed on the articles surrogate id, which does not survive
++	// a retention purge and re-fetch of the same feed entry. read_state_history
++	// records the same fact against the article's stable (feed_url, guid)
++	// identity so "read" is not forgotten when a purged article is re-ingested.
++	`CREATE TABLE IF NOT EXISTS articles.read_state_history (
++		user_did TEXT NOT NULL,
++		feed_url TEXT NOT NULL,
++		guid TEXT NOT NULL,
++		is_read BOOLEAN NOT NULL DEFAULT 0,
++		read_at DATETIME,
++		PRIMARY KEY (user_did, feed_url, guid)
++	)`,
++
  	`CREATE TABLE IF NOT EXISTS articles.annotations (
  		id INTEGER PRIMARY KEY AUTOINCREMENT,
  		uri TEXT NOT NULL UNIQUE,

modified internal/db/retention.go +21 -1 Viewed

@@ -83,13 +83,33 @@ func (s *Store) deleteInBatches(ctx context.Context, query string) (int64, error
83	83
84	84	// PurgeExpiredArticles deletes articles older than maxAgeDays — by published
85	85	// date, falling back to fetched_at for undated items — together with their
86		-// read-state rows. It returns the number of articles removed.
	86	+// read-state rows, which are first archived to read_state_history so a
	87	+// re-ingested article is not resurrected as unread. It returns the number of
	88	+// articles removed.
87	89	func (s *Store) PurgeExpiredArticles(ctx context.Context, maxAgeDays int) (int64, error) {
88	90	if maxAgeDays <= 0 {
89	91	return 0, nil
90	92	}
91	93	cutoff := time.Now().AddDate(0, 0, -maxAgeDays)
92	94	const stale = `(published IS NOT NULL AND published < ?1) OR (published IS NULL AND fetched_at < ?1)`
	95	+ const staleA = `(a.published IS NOT NULL AND a.published < ?1) OR (a.published IS NULL AND a.fetched_at < ?1)`
	96	+
	97	+ // Archive read state against the article's stable identity before dropping
	98	+ // it. Feeds keep serving entries past the retention cutoff, so a purged
	99	+ // article is routinely re-ingested under a fresh surrogate id; without this
	100	+ // the user's "read" mark is lost and the backlog resurfaces as unread.
	101	+ if _, err := s.db.ExecContext(ctx,
	102	+ `INSERT INTO articles.read_state_history (user_did, feed_url, guid, is_read, read_at)
	103	+ SELECT r.user_did, a.feed_url, a.guid, r.is_read, r.read_at
	104	+ FROM articles.read_state r
	105	+ JOIN articles.articles a ON a.id = r.article_id
	106	+ WHERE `+staleA+`
	107	+ ON CONFLICT(user_did, feed_url, guid) DO UPDATE SET
	108	+ is_read = excluded.is_read, read_at = excluded.read_at`,
	109	+ cutoff,
	110	+ ); err != nil {
	111	+ return 0, fmt.Errorf("archive read state of old articles: %w", err)
	112	+ }
93	113
94	114	if _, err := s.db.ExecContext(ctx,
95	115	`DELETE FROM articles.read_state WHERE article_id IN (SELECT id FROM articles.articles WHERE `+stale+`)`,

  // PurgeExpiredArticles deletes articles older than maxAgeDays — by published
  // date, falling back to fetched_at for undated items — together with their
--// read-state rows. It returns the number of articles removed.
++// read-state rows, which are first archived to read_state_history so a
++// re-ingested article is not resurrected as unread. It returns the number of
++// articles removed.
  func (s *Store) PurgeExpiredArticles(ctx context.Context, maxAgeDays int) (int64, error) {
  	if maxAgeDays <= 0 {
  		return 0, nil
  	}
  	cutoff := time.Now().AddDate(0, 0, -maxAgeDays)
  	const stale = `(published IS NOT NULL AND published < ?1) OR (published IS NULL AND fetched_at < ?1)`
++	const staleA = `(a.published IS NOT NULL AND a.published < ?1) OR (a.published IS NULL AND a.fetched_at < ?1)`
++
++	// Archive read state against the article's stable identity before dropping
++	// it. Feeds keep serving entries past the retention cutoff, so a purged
++	// article is routinely re-ingested under a fresh surrogate id; without this
++	// the user's "read" mark is lost and the backlog resurfaces as unread.
++	if _, err := s.db.ExecContext(ctx,
++		`INSERT INTO articles.read_state_history (user_did, feed_url, guid, is_read, read_at)
++		SELECT r.user_did, a.feed_url, a.guid, r.is_read, r.read_at
++		FROM articles.read_state r
++		JOIN articles.articles a ON a.id = r.article_id
++		WHERE `+staleA+`
++		ON CONFLICT(user_did, feed_url, guid) DO UPDATE SET
++			is_read = excluded.is_read, read_at = excluded.read_at`,
++		cutoff,
++	); err != nil {
++		return 0, fmt.Errorf("archive read state of old articles: %w", err)
++	}
  	if _, err := s.db.ExecContext(ctx,
  		`DELETE FROM articles.read_state WHERE article_id IN (SELECT id FROM articles.articles WHERE `+stale+`)`,

modified internal/db/retention_test.go +62 -0 Viewed

@@ -6,6 +6,8 @@ import (
6	6	"time"
7	7
8	8	"gotest.tools/v3/assert"
	9	+
	10	+ "pkg.rbrt.fr/glean/internal/feed"
9	11	)
10	12
11	13	func seedRetentionFixtures(t testing.T, ctx context.Context, dbs Store) {
@@ -131,3 +133,63 @@ func TestRunMaintenance_AppliesArticleRetention(t *testing.T) {
131	133	assert.NilError(t, dbs.RunMaintenance(ctx, 90, 30))
132	134	assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles`))
133	135	}
	136	+
	137	+// A feed keeps serving entries past the retention cutoff, so a purged article
	138	+// comes back under a fresh surrogate id. Read state must come back with it.
	139	+func TestPurgeThenReingest_KeepsArticleRead(t *testing.T) {
	140	+ ctx := context.Background()
	141	+ dbs := setupTestDB(t)
	142	+ sqlDB := dbs.SQLDB()
	143	+
	144	+ const did = "did:test:known"
	145	+ _, err := sqlDB.ExecContext(ctx, `INSERT INTO users (did) VALUES (?)`, did)
	146	+ assert.NilError(t, err)
	147	+
	148	+ published := time.Now().AddDate(0, 0, -60)
	149	+ _, err = sqlDB.ExecContext(ctx, `
	150	+ INSERT INTO articles.articles (id, feed_url, guid, title, published)
	151	+ VALUES (1, 'https://f.example/rss', 'weekly-92', 't', ?)`, published)
	152	+ assert.NilError(t, err)
	153	+ assert.NilError(t, dbs.Articles.MarkArticleRead(ctx, did, 1))
	154	+
	155	+ n, err := dbs.PurgeExpiredArticles(ctx, 30)
	156	+ assert.NilError(t, err)
	157	+ assert.Equal(t, int64(1), n)
	158	+
	159	+ // The feed still lists it; ingest with no cutoff configured re-adds it.
	160	+ assert.NilError(t, dbs.Articles.BatchUpsertArticles(ctx, []feed.Article{{
	161	+ FeedURL: "https://f.example/rss",
	162	+ GUID: "weekly-92",
	163	+ Title: "t",
	164	+ Published: published,
	165	+ }}))
	166	+
	167	+ var id int64
	168	+ var isRead bool
	169	+ err = sqlDB.QueryRowContext(ctx, `
	170	+ SELECT a.id, COALESCE(r.is_read, 0)
	171	+ FROM articles.articles a
	172	+ LEFT JOIN articles.read_state r ON r.user_did = ? AND r.article_id = a.id
	173	+ WHERE a.guid = 'weekly-92'`, did).Scan(&id, &isRead)
	174	+ assert.NilError(t, err)
	175	+ assert.Assert(t, id != 1, "re-ingested article should have a new surrogate id")
	176	+ assert.Equal(t, true, isRead, "read state must survive purge and re-ingest")
	177	+}
	178	+
	179	+func TestBatchUpsertArticles_SkipsArticlesPastRetention(t *testing.T) {
	180	+ ctx := context.Background()
	181	+ dbs := setupTestDB(t)
	182	+ dbs.Articles.SetRetentionDays(30)
	183	+
	184	+ now := time.Now()
	185	+ assert.NilError(t, dbs.Articles.BatchUpsertArticles(ctx, []feed.Article{
	186	+ {FeedURL: "https://f.example/rss", GUID: "old", Title: "t", Published: now.AddDate(0, 0, -60)},
	187	+ {FeedURL: "https://f.example/rss", GUID: "fresh", Title: "t", Published: now.AddDate(0, 0, -5)},
	188	+ {FeedURL: "https://f.example/rss", GUID: "undated", Title: "t"},
	189	+ }))
	190	+
	191	+ assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'old'`))
	192	+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'fresh'`))
	193	+ // Undated entries have no age to judge, so they are still ingested.
	194	+ assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'undated'`))
	195	+}

  	"time"
  	"gotest.tools/v3/assert"
++
++	"pkg.rbrt.fr/glean/internal/feed"
  )
  func seedRetentionFixtures(t *testing.T, ctx context.Context, dbs *Store) {
  	assert.NilError(t, dbs.RunMaintenance(ctx, 90, 30))
  	assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles`))
  }
++
++// A feed keeps serving entries past the retention cutoff, so a purged article
++// comes back under a fresh surrogate id. Read state must come back with it.
++func TestPurgeThenReingest_KeepsArticleRead(t *testing.T) {
++	ctx := context.Background()
++	dbs := setupTestDB(t)
++	sqlDB := dbs.SQLDB()
++
++	const did = "did:test:known"
++	_, err := sqlDB.ExecContext(ctx, `INSERT INTO users (did) VALUES (?)`, did)
++	assert.NilError(t, err)
++
++	published := time.Now().AddDate(0, 0, -60)
++	_, err = sqlDB.ExecContext(ctx, `
++		INSERT INTO articles.articles (id, feed_url, guid, title, published)
++		VALUES (1, 'https://f.example/rss', 'weekly-92', 't', ?)`, published)
++	assert.NilError(t, err)
++	assert.NilError(t, dbs.Articles.MarkArticleRead(ctx, did, 1))
++
++	n, err := dbs.PurgeExpiredArticles(ctx, 30)
++	assert.NilError(t, err)
++	assert.Equal(t, int64(1), n)
++
++	// The feed still lists it; ingest with no cutoff configured re-adds it.
++	assert.NilError(t, dbs.Articles.BatchUpsertArticles(ctx, []feed.Article{{
++		FeedURL:   "https://f.example/rss",
++		GUID:      "weekly-92",
++		Title:     "t",
++		Published: published,
++	}}))
++
++	var id int64
++	var isRead bool
++	err = sqlDB.QueryRowContext(ctx, `
++		SELECT a.id, COALESCE(r.is_read, 0)
++		FROM articles.articles a
++		LEFT JOIN articles.read_state r ON r.user_did = ? AND r.article_id = a.id
++		WHERE a.guid = 'weekly-92'`, did).Scan(&id, &isRead)
++	assert.NilError(t, err)
++	assert.Assert(t, id != 1, "re-ingested article should have a new surrogate id")
++	assert.Equal(t, true, isRead, "read state must survive purge and re-ingest")
++}
++
++func TestBatchUpsertArticles_SkipsArticlesPastRetention(t *testing.T) {
++	ctx := context.Background()
++	dbs := setupTestDB(t)
++	dbs.Articles.SetRetentionDays(30)
++
++	now := time.Now()
++	assert.NilError(t, dbs.Articles.BatchUpsertArticles(ctx, []feed.Article{
++		{FeedURL: "https://f.example/rss", GUID: "old", Title: "t", Published: now.AddDate(0, 0, -60)},
++		{FeedURL: "https://f.example/rss", GUID: "fresh", Title: "t", Published: now.AddDate(0, 0, -5)},
++		{FeedURL: "https://f.example/rss", GUID: "undated", Title: "t"},
++	}))
++
++	assert.Equal(t, int64(0), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'old'`))
++	assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'fresh'`))
++	// Undated entries have no age to judge, so they are still ingested.
++	assert.Equal(t, int64(1), count(t, ctx, dbs, `SELECT COUNT(*) FROM articles.articles WHERE guid = 'undated'`))
++}

modified main.go +3 -0 Viewed

@@ -68,6 +68,9 @@ func main() {
68	68	os.Exit(1)
69	69	}
70	70	defer dbs.Close()
	71	+ // Ingest must use the same window as the purge, or every fetch re-adds the
	72	+ // articles the last purge deleted.
	73	+ dbs.Articles.SetRetentionDays(*articleRetentionDays)
71	74
72	75	clientID := envOr("GLEAN_OAUTH_CLIENT_ID", "")
73	76	frontendURL := envOr("GLEAN_FRONTEND_URL", "")

  		os.Exit(1)
  	}
  	defer dbs.Close()
++	// Ingest must use the same window as the purge, or every fetch re-adds the
++	// articles the last purge deleted.
++	dbs.Articles.SetRetentionDays(*articleRetentionDays)
  	clientID := envOr("GLEAN_OAUTH_CLIENT_ID", "")
  	frontendURL := envOr("GLEAN_FRONTEND_URL", "")