1use std::fmt;
2
3use anyhow::{Context, Result};
4use chrono::{DateTime, Duration, Utc};
5use serde::{Deserialize, Serialize};
6
7use super::{Database, Issue};
8
9pub const HYDRATION_RESOURCES: [HydrationResource; 4] = [
10 HydrationResource::Details,
11 HydrationResource::Labels,
12 HydrationResource::Relations,
13 HydrationResource::Comments,
14];
15
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
17pub enum HydrationResource {
18 Details,
19 Labels,
20 Relations,
21 Comments,
22}
23
24impl HydrationResource {
25 pub fn as_str(self) -> &'static str {
26 match self {
27 Self::Details => "details",
28 Self::Labels => "labels",
29 Self::Relations => "relations",
30 Self::Comments => "comments",
31 }
32 }
33
34 fn parse(value: &str) -> Result<Self> {
35 match value {
36 "details" => Ok(Self::Details),
37 "labels" => Ok(Self::Labels),
38 "relations" => Ok(Self::Relations),
39 "comments" => Ok(Self::Comments),
40 _ => anyhow::bail!("unknown hydration resource '{value}'"),
41 }
42 }
43}
44
45impl fmt::Display for HydrationResource {
46 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
47 f.write_str(self.as_str())
48 }
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
52pub enum HydrationStatus {
53 Pending,
54 Running,
55 Hydrated,
56 Partial,
57 Retryable,
58 PermissionDenied,
59 Unavailable,
60}
61
62impl HydrationStatus {
63 pub fn as_str(self) -> &'static str {
64 match self {
65 Self::Pending => "pending",
66 Self::Running => "running",
67 Self::Hydrated => "hydrated",
68 Self::Partial => "partial",
69 Self::Retryable => "retryable",
70 Self::PermissionDenied => "permission_denied",
71 Self::Unavailable => "unavailable",
72 }
73 }
74
75 fn parse(value: &str) -> Result<Self> {
76 match value {
77 "pending" => Ok(Self::Pending),
78 "running" => Ok(Self::Running),
79 "hydrated" => Ok(Self::Hydrated),
80 "partial" => Ok(Self::Partial),
81 "retryable" => Ok(Self::Retryable),
82 "permission_denied" => Ok(Self::PermissionDenied),
83 "unavailable" => Ok(Self::Unavailable),
84 _ => anyhow::bail!("unknown hydration status '{value}'"),
85 }
86 }
87}
88
89#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
90pub enum HydrationPolicy {
91 OpenOnly,
92 OpenAndRecent,
93 All,
94}
95
96#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
97pub enum HydrationMode {
98 IfNeeded,
99 ForceRefresh,
100}
101
102#[derive(Debug, Clone, PartialEq, Eq)]
103pub struct HydrationResourceState {
104 pub resource: HydrationResource,
105 pub status: HydrationStatus,
106 pub source_updated_at: String,
107 pub last_attempted_at: Option<String>,
108 pub hydrated_at: Option<String>,
109 pub attempt_count: u32,
110 pub next_retry_at: Option<String>,
111 pub last_error: Option<String>,
112}
113
114#[derive(Debug, Clone, PartialEq, Eq)]
115pub struct IssueHydrationState {
116 pub issue_id: String,
117 pub status: HydrationStatus,
118 pub resources: Vec<HydrationResourceState>,
119}
120
121#[derive(Debug, Clone, PartialEq, Eq)]
122pub struct HydrationCandidate {
123 pub id: String,
124 pub identifier: String,
125 pub updated_at: String,
126 pub state_type: String,
127}
128
129#[derive(Debug, Clone)]
130pub struct IssueIndexEntry {
131 pub id: String,
132 pub identifier: String,
133 pub team_key: String,
134 pub title: String,
135 pub state_name: String,
136 pub state_type: String,
137 pub created_at: String,
138 pub updated_at: String,
139 pub archived_at: Option<String>,
140 pub url: String,
141}
142
143#[derive(Debug, Clone, Copy, PartialEq, Eq)]
144pub enum IndexUpsertOutcome {
145 Inserted,
146 Updated,
147 Unchanged,
148}
149
150impl Database {
151 pub fn upsert_issue_index(
154 &self,
155 issue: &IssueIndexEntry,
156 workspace_id: &str,
157 sync_token: &str,
158 ) -> Result<IndexUpsertOutcome> {
159 self.with_conn(|conn| {
160 let existing = conn
161 .query_row(
162 "SELECT identifier, team_key, title, state_name, state_type,
163 created_at, updated_at, archived_at, url, description
164 FROM issues WHERE id = ?1",
165 rusqlite::params![issue.id],
166 |row| {
167 Ok((
168 row.get::<_, String>(0)?,
169 row.get::<_, String>(1)?,
170 row.get::<_, String>(2)?,
171 row.get::<_, String>(3)?,
172 row.get::<_, String>(4)?,
173 row.get::<_, String>(5)?,
174 row.get::<_, String>(6)?,
175 row.get::<_, Option<String>>(7)?,
176 row.get::<_, String>(8)?,
177 row.get::<_, Option<String>>(9)?,
178 ))
179 },
180 )
181 .optional()?;
182
183 let outcome = match &existing {
184 None => IndexUpsertOutcome::Inserted,
185 Some(current)
186 if current.0 == issue.identifier
187 && current.1 == issue.team_key
188 && current.2 == issue.title
189 && current.3 == issue.state_name
190 && current.4 == issue.state_type
191 && current.5 == issue.created_at
192 && current.6 == issue.updated_at
193 && current.7 == issue.archived_at
194 && current.8 == issue.url
195 =>
196 {
197 IndexUpsertOutcome::Unchanged
198 }
199 Some(_) => IndexUpsertOutcome::Updated,
200 };
201 let preserved_description = existing
202 .as_ref()
203 .and_then(|current| current.9.as_deref());
204 let embedding_content_hash =
205 crate::embedding::issue_content_hash(&issue.title, preserved_description);
206
207 let tx = conn.unchecked_transaction()?;
208 tx.execute(
209 "INSERT INTO issues (
210 id, identifier, team_key, title, description, state_name,
211 state_type, priority, assignee_name, project_name,
212 labels_json, created_at, updated_at, content_hash, synced_at,
213 url, branch_name, workspace_id, archived_at, sync_token,
214 embedding_content_hash
215 ) VALUES (
216 ?1, ?2, ?3, ?4, NULL, ?5, ?6, 0, NULL, NULL, '[]', ?7,
217 ?8, '', datetime('now'), ?9, NULL, ?10, ?11, ?12, ?13
218 ) ON CONFLICT(id) DO UPDATE SET
219 identifier=excluded.identifier,
220 team_key=excluded.team_key,
221 title=excluded.title,
222 state_name=excluded.state_name,
223 state_type=excluded.state_type,
224 created_at=excluded.created_at,
225 updated_at=excluded.updated_at,
226 archived_at=excluded.archived_at,
227 url=excluded.url,
228 workspace_id=excluded.workspace_id,
229 sync_token=excluded.sync_token,
230 embedding_content_hash=excluded.embedding_content_hash,
231 synced_at=datetime('now')",
232 rusqlite::params![
233 issue.id,
234 issue.identifier,
235 issue.team_key,
236 issue.title,
237 issue.state_name,
238 issue.state_type,
239 issue.created_at,
240 issue.updated_at,
241 issue.url,
242 workspace_id,
243 issue.archived_at,
244 sync_token,
245 embedding_content_hash,
246 ],
247 )?;
248
249 if outcome != IndexUpsertOutcome::Unchanged {
250 for resource in HYDRATION_RESOURCES {
251 tx.execute(
252 "INSERT INTO issue_hydration_state (
253 workspace_id, issue_id, resource, status,
254 source_updated_at, queue_reason, index_sync_token, attempt_count,
255 next_retry_at, last_error
256 ) VALUES (?1, ?2, ?3, 'pending', ?4, 'index_changed', ?5, 0, NULL, NULL)
257 ON CONFLICT(workspace_id, issue_id, resource) DO UPDATE SET
258 status=CASE
259 WHEN issue_hydration_state.source_updated_at <> excluded.source_updated_at
260 THEN 'pending' ELSE issue_hydration_state.status END,
261 source_updated_at=excluded.source_updated_at,
262 queue_reason=CASE
263 WHEN issue_hydration_state.source_updated_at <> excluded.source_updated_at
264 THEN 'index_changed' ELSE issue_hydration_state.queue_reason END,
265 index_sync_token=CASE
266 WHEN issue_hydration_state.source_updated_at <> excluded.source_updated_at
267 THEN excluded.index_sync_token ELSE issue_hydration_state.index_sync_token END,
268 attempt_count=CASE
269 WHEN issue_hydration_state.source_updated_at <> excluded.source_updated_at
270 THEN 0 ELSE issue_hydration_state.attempt_count END,
271 next_retry_at=CASE
272 WHEN issue_hydration_state.source_updated_at <> excluded.source_updated_at
273 THEN NULL ELSE issue_hydration_state.next_retry_at END,
274 last_error=CASE
275 WHEN issue_hydration_state.source_updated_at <> excluded.source_updated_at
276 THEN NULL ELSE issue_hydration_state.last_error END",
277 rusqlite::params![
278 workspace_id,
279 issue.id,
280 resource.as_str(),
281 issue.updated_at,
282 sync_token,
283 ],
284 )?;
285 }
286 }
287 tx.commit()?;
288 Ok(outcome)
289 })
290 }
291
292 pub fn mark_hydration_running(
293 &self,
294 workspace_id: &str,
295 issue_id: &str,
296 resource: HydrationResource,
297 attempted_at: &str,
298 ) -> Result<u32> {
299 self.with_conn(|conn| {
300 conn.execute(
301 "UPDATE issue_hydration_state
302 SET status='running', last_attempted_at=?4,
303 attempt_count=attempt_count + 1
304 WHERE workspace_id=?1 AND issue_id=?2 AND resource=?3",
305 rusqlite::params![workspace_id, issue_id, resource.as_str(), attempted_at],
306 )?;
307 let attempts = conn.query_row(
308 "SELECT attempt_count FROM issue_hydration_state
309 WHERE workspace_id=?1 AND issue_id=?2 AND resource=?3",
310 rusqlite::params![workspace_id, issue_id, resource.as_str()],
311 |row| row.get::<_, u32>(0),
312 )?;
313 Ok(attempts)
314 })
315 }
316
317 pub fn requeue_issue_hydration(
318 &self,
319 workspace_id: &str,
320 issue_id: &str,
321 reason: &str,
322 ) -> Result<()> {
323 let issue = self
324 .get_issue(issue_id)?
325 .with_context(|| format!("issue '{issue_id}' not found"))?;
326 self.ensure_hydration_state_for_issue(workspace_id, &issue, reason)?;
327 self.with_conn(|conn| {
328 conn.execute(
329 "UPDATE issue_hydration_state
330 SET status='pending', queue_reason=?3, next_retry_at=NULL,
331 last_error=NULL
332 WHERE workspace_id=?1 AND issue_id=?2",
333 rusqlite::params![workspace_id, issue.id, reason],
334 )?;
335 Ok(())
336 })
337 }
338
339 pub fn requeue_team_hydration(
340 &self,
341 workspace_id: &str,
342 team_key: &str,
343 reason: &str,
344 ) -> Result<usize> {
345 self.with_conn(|conn| {
346 Ok(conn.execute(
347 "UPDATE issue_hydration_state
348 SET status='pending', queue_reason=?3, next_retry_at=NULL,
349 last_error=NULL
350 WHERE workspace_id=?1 AND issue_id IN (
351 SELECT id FROM issues WHERE workspace_id=?1 AND team_key=?2
352 )",
353 rusqlite::params![workspace_id, team_key, reason],
354 )?)
355 })
356 }
357
358 pub fn requeue_team_retryable_comments(
359 &self,
360 workspace_id: &str,
361 team_key: &str,
362 ) -> Result<usize> {
363 self.with_conn(|conn| {
364 Ok(conn.execute(
365 "UPDATE issue_hydration_state
366 SET status='pending', queue_reason='legacy_retry', next_retry_at=NULL
367 WHERE workspace_id=?1 AND resource='comments'
368 AND status='retryable' AND issue_id IN (
369 SELECT id FROM issues WHERE workspace_id=?1 AND team_key=?2
370 )",
371 rusqlite::params![workspace_id, team_key],
372 )?)
373 })
374 }
375
376 pub fn mark_hydration_complete(
377 &self,
378 workspace_id: &str,
379 issue_id: &str,
380 resource: HydrationResource,
381 source_updated_at: &str,
382 hydrated_at: &str,
383 ) -> Result<()> {
384 self.with_conn(|conn| {
385 conn.execute(
386 "UPDATE issue_hydration_state
387 SET status='hydrated', source_updated_at=?4, hydrated_at=?5,
388 next_retry_at=NULL, last_error=NULL, queue_reason='complete'
389 WHERE workspace_id=?1 AND issue_id=?2 AND resource=?3",
390 rusqlite::params![
391 workspace_id,
392 issue_id,
393 resource.as_str(),
394 source_updated_at,
395 hydrated_at,
396 ],
397 )?;
398 Ok(())
399 })
400 }
401
402 pub fn mark_hydration_failed(
403 &self,
404 workspace_id: &str,
405 issue_id: &str,
406 resource: HydrationResource,
407 status: HydrationStatus,
408 next_retry_at: Option<&str>,
409 error: &str,
410 ) -> Result<()> {
411 self.with_conn(|conn| {
412 conn.execute(
413 "UPDATE issue_hydration_state
414 SET status=?4, next_retry_at=?5, last_error=?6,
415 queue_reason=CASE WHEN ?4='retryable' THEN 'retry' ELSE queue_reason END
416 WHERE workspace_id=?1 AND issue_id=?2 AND resource=?3",
417 rusqlite::params![
418 workspace_id,
419 issue_id,
420 resource.as_str(),
421 status.as_str(),
422 next_retry_at,
423 error,
424 ],
425 )?;
426 Ok(())
427 })
428 }
429
430 pub fn get_issue_hydration_state(
431 &self,
432 workspace_id: &str,
433 issue_id: &str,
434 ) -> Result<IssueHydrationState> {
435 self.with_conn(|conn| {
436 let mut stmt = conn.prepare(
437 "SELECT resource, status, source_updated_at, last_attempted_at,
438 hydrated_at, attempt_count, next_retry_at, last_error
439 FROM issue_hydration_state
440 WHERE workspace_id=?1 AND issue_id=?2
441 ORDER BY CASE resource
442 WHEN 'details' THEN 1 WHEN 'labels' THEN 2
443 WHEN 'relations' THEN 3 ELSE 4 END",
444 )?;
445 let rows = stmt.query_map(rusqlite::params![workspace_id, issue_id], |row| {
446 Ok((
447 row.get::<_, String>(0)?,
448 row.get::<_, String>(1)?,
449 row.get::<_, String>(2)?,
450 row.get::<_, Option<String>>(3)?,
451 row.get::<_, Option<String>>(4)?,
452 row.get::<_, u32>(5)?,
453 row.get::<_, Option<String>>(6)?,
454 row.get::<_, Option<String>>(7)?,
455 ))
456 })?;
457 let resources = rows
458 .map(|row| {
459 let row = row?;
460 Ok(HydrationResourceState {
461 resource: HydrationResource::parse(&row.0)?,
462 status: HydrationStatus::parse(&row.1)?,
463 source_updated_at: row.2,
464 last_attempted_at: row.3,
465 hydrated_at: row.4,
466 attempt_count: row.5,
467 next_retry_at: row.6,
468 last_error: row.7,
469 })
470 })
471 .collect::<Result<Vec<_>>>()?;
472 if resources.is_empty() {
473 anyhow::bail!("no hydration state for issue '{issue_id}'");
474 }
475 let status = aggregate_hydration_status(&resources);
476 Ok(IssueHydrationState {
477 issue_id: issue_id.to_string(),
478 status,
479 resources,
480 })
481 })
482 }
483
484 pub fn queue_stale_comment_hydration(
485 &self,
486 workspace_id: &str,
487 team_key: &str,
488 policy: HydrationPolicy,
489 stale_before: &str,
490 recent_after: &str,
491 ) -> Result<usize> {
492 self.with_conn(|conn| {
493 let policy_filter = match policy {
494 HydrationPolicy::OpenOnly => "i.state_type NOT IN ('completed', 'canceled')",
495 HydrationPolicy::OpenAndRecent => {
496 "(i.state_type NOT IN ('completed', 'canceled') OR julianday(i.updated_at) >= julianday(?5))"
497 }
498 HydrationPolicy::All => "1=1",
499 };
500 let sql = format!(
501 "UPDATE issue_hydration_state AS h
502 SET status='pending', queue_reason='comment_refresh'
503 WHERE h.workspace_id=?1 AND h.resource='comments'
504 AND h.status='hydrated'
505 AND julianday(h.hydrated_at) <= julianday(?3)
506 AND EXISTS (
507 SELECT 1 FROM issues i
508 WHERE i.id=h.issue_id AND i.workspace_id=?1
509 AND i.team_key=?2 AND {policy_filter}
510 )"
511 );
512 let changed = match policy {
513 HydrationPolicy::OpenAndRecent => conn.execute(
514 &sql,
515 rusqlite::params![workspace_id, team_key, stale_before, 0, recent_after],
516 )?,
517 _ => conn.execute(
518 &sql,
519 rusqlite::params![workspace_id, team_key, stale_before],
520 )?,
521 };
522 Ok(changed)
523 })
524 }
525
526 pub fn queue_stale_issue_comment_hydration(
527 &self,
528 workspace_id: &str,
529 issue_id: &str,
530 stale_before: &str,
531 ) -> Result<usize> {
532 self.with_conn(|conn| {
533 Ok(conn.execute(
534 "UPDATE issue_hydration_state
535 SET status='pending', queue_reason='comment_refresh'
536 WHERE workspace_id=?1 AND issue_id=?2 AND resource='comments'
537 AND status='hydrated'
538 AND julianday(hydrated_at) <= julianday(?3)",
539 rusqlite::params![workspace_id, issue_id, stale_before],
540 )?)
541 })
542 }
543
544 pub fn list_hydration_candidates(
545 &self,
546 workspace_id: &str,
547 team_key: &str,
548 limit: usize,
549 policy: HydrationPolicy,
550 now: &str,
551 recent_after: &str,
552 ) -> Result<Vec<HydrationCandidate>> {
553 self.with_conn(|conn| {
554 let policy_filter = match policy {
555 HydrationPolicy::OpenOnly => "i.state_type NOT IN ('completed', 'canceled')",
556 HydrationPolicy::OpenAndRecent => {
557 "(i.state_type NOT IN ('completed', 'canceled') OR julianday(i.updated_at) >= julianday(?4))"
558 }
559 HydrationPolicy::All => "1=1",
560 };
561 let sql = format!(
562 "SELECT i.id, i.identifier, i.updated_at, i.state_type
563 FROM issues i
564 WHERE i.workspace_id=?1 AND i.team_key=?2
565 AND {policy_filter}
566 AND EXISTS (
567 SELECT 1 FROM issue_hydration_state h
568 WHERE h.workspace_id=i.workspace_id AND h.issue_id=i.id
569 AND (h.status='pending' OR (
570 h.status='retryable' AND (
571 h.next_retry_at IS NULL OR julianday(h.next_retry_at) <= julianday(?3)
572 )
573 ))
574 )
575 ORDER BY
576 CASE
577 WHEN i.state_type NOT IN ('completed', 'canceled') AND EXISTS (
578 SELECT 1 FROM issue_hydration_state h
579 WHERE h.workspace_id=i.workspace_id AND h.issue_id=i.id
580 AND h.status='pending' AND h.queue_reason='index_changed'
581 AND h.index_sync_token=(
582 SELECT latest.sync_token FROM sync_family_state latest
583 WHERE latest.workspace_id=i.workspace_id
584 AND latest.team_key=i.team_key
585 AND latest.family='issue index'
586 )
587 ) THEN 2
588 WHEN EXISTS (
589 SELECT 1 FROM issue_hydration_state h
590 WHERE h.workspace_id=i.workspace_id AND h.issue_id=i.id
591 AND h.status='retryable'
592 AND (h.next_retry_at IS NULL OR julianday(h.next_retry_at) <= julianday(?3))
593 ) THEN 3
594 WHEN i.state_type NOT IN ('completed', 'canceled') THEN 4
595 WHEN julianday(i.updated_at) >= julianday(?4) THEN 5
596 ELSE 6
597 END,
598 julianday(i.updated_at) DESC, i.identifier ASC
599 LIMIT ?5"
600 );
601 let mut stmt = conn.prepare(&sql)?;
602 let rows = stmt.query_map(
603 rusqlite::params![workspace_id, team_key, now, recent_after, limit as i64],
604 |row| {
605 Ok(HydrationCandidate {
606 id: row.get(0)?,
607 identifier: row.get(1)?,
608 updated_at: row.get(2)?,
609 state_type: row.get(3)?,
610 })
611 },
612 )?;
613 Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
614 })
615 }
616
617 pub fn ensure_hydration_state_for_issue(
618 &self,
619 workspace_id: &str,
620 issue: &Issue,
621 reason: &str,
622 ) -> Result<()> {
623 self.with_conn(|conn| {
624 for resource in HYDRATION_RESOURCES {
625 conn.execute(
626 "INSERT INTO issue_hydration_state (
627 workspace_id, issue_id, resource, status,
628 source_updated_at, queue_reason
629 ) VALUES (?1, ?2, ?3, 'pending', ?4, ?5)
630 ON CONFLICT(workspace_id, issue_id, resource) DO NOTHING",
631 rusqlite::params![
632 workspace_id,
633 issue.id,
634 resource.as_str(),
635 issue.updated_at,
636 reason,
637 ],
638 )?;
639 }
640 Ok(())
641 })
642 }
643}
644
645fn aggregate_hydration_status(resources: &[HydrationResourceState]) -> HydrationStatus {
646 if resources
647 .iter()
648 .all(|state| state.status == HydrationStatus::Hydrated)
649 {
650 return HydrationStatus::Hydrated;
651 }
652 if resources
653 .iter()
654 .any(|state| state.status == HydrationStatus::Running)
655 {
656 return HydrationStatus::Running;
657 }
658 if resources
659 .iter()
660 .any(|state| state.status == HydrationStatus::Pending)
661 {
662 return HydrationStatus::Pending;
663 }
664 if resources
665 .iter()
666 .any(|state| state.status == HydrationStatus::Retryable)
667 {
668 return HydrationStatus::Retryable;
669 }
670 let hydrated = resources
671 .iter()
672 .any(|state| state.status == HydrationStatus::Hydrated);
673 if hydrated {
674 HydrationStatus::Partial
675 } else if resources
676 .iter()
677 .all(|state| state.status == HydrationStatus::PermissionDenied)
678 {
679 HydrationStatus::PermissionDenied
680 } else {
681 HydrationStatus::Unavailable
682 }
683}
684
685pub fn recent_cutoff(now: DateTime<Utc>) -> String {
686 (now - Duration::days(30)).to_rfc3339()
687}
688
689pub fn comment_refresh_cutoff(now: DateTime<Utc>) -> String {
690 (now - Duration::minutes(15)).to_rfc3339()
691}
692
693trait OptionalRow<T> {
694 fn optional(self) -> rusqlite::Result<Option<T>>;
695}
696
697impl<T> OptionalRow<T> for rusqlite::Result<T> {
698 fn optional(self) -> rusqlite::Result<Option<T>> {
699 match self {
700 Ok(value) => Ok(Some(value)),
701 Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
702 Err(error) => Err(error),
703 }
704 }
705}
706
707pub fn parse_timestamp(value: &str) -> Result<DateTime<Utc>> {
708 DateTime::parse_from_rfc3339(value)
709 .map(|value| value.with_timezone(&Utc))
710 .with_context(|| format!("invalid RFC3339 timestamp '{value}'"))
711}
712
713#[cfg(test)]
714mod tests {
715 use super::*;
716 use crate::db::test_helpers::make_issue;
717
718 #[test]
719 fn index_upsert_preserves_hydrated_fields_and_requeues_on_change() {
720 let (db, _dir) = crate::db::test_helpers::test_db();
721 let mut rich = make_issue("CUT-1", "CUT");
722 rich.id = "issue-1".into();
723 rich.description = Some("hydrated description".into());
724 rich.labels_json = r#"["bug"]"#.into();
725 db.upsert_issue(&rich).unwrap();
726 db.ensure_hydration_state_for_issue("default", &rich, "test")
727 .unwrap();
728
729 let outcome = db
730 .upsert_issue_index(
731 &IssueIndexEntry {
732 id: rich.id.clone(),
733 identifier: rich.identifier.clone(),
734 team_key: rich.team_key.clone(),
735 title: "new index title".into(),
736 state_name: "In Progress".into(),
737 state_type: "started".into(),
738 created_at: rich.created_at.clone(),
739 updated_at: "2026-01-03T00:00:00Z".into(),
740 archived_at: None,
741 url: rich.url.clone(),
742 },
743 "default",
744 "run-1",
745 )
746 .unwrap();
747
748 assert_eq!(outcome, IndexUpsertOutcome::Updated);
749 let stored = db.get_issue("issue-1").unwrap().unwrap();
750 assert_eq!(stored.title, "new index title");
751 assert_eq!(stored.description.as_deref(), Some("hydrated description"));
752 assert_eq!(stored.labels_json, r#"["bug"]"#);
753 let state = db.get_issue_hydration_state("default", "issue-1").unwrap();
754 assert!(state
755 .resources
756 .iter()
757 .all(|resource| resource.status == HydrationStatus::Pending));
758 }
759
760 #[test]
761 fn open_changed_issues_are_prioritized_over_old_completed_issues() {
762 let (db, _dir) = crate::db::test_helpers::test_db();
763 let mut completed = make_issue("CUT-1", "CUT");
764 completed.id = "completed".into();
765 completed.state_type = "completed".into();
766 completed.updated_at = "2020-01-01T00:00:00Z".into();
767 db.upsert_issue(&completed).unwrap();
768 db.ensure_hydration_state_for_issue("default", &completed, "initial")
769 .unwrap();
770
771 let mut open = make_issue("CUT-2", "CUT");
772 open.id = "open".into();
773 open.updated_at = "2025-01-01T00:00:00Z".into();
774 db.upsert_issue(&open).unwrap();
775 db.ensure_hydration_state_for_issue("default", &open, "index_changed")
776 .unwrap();
777
778 let candidates = db
779 .list_hydration_candidates(
780 "default",
781 "CUT",
782 10,
783 HydrationPolicy::All,
784 "2026-01-01T00:00:00Z",
785 "2025-12-01T00:00:00Z",
786 )
787 .unwrap();
788 assert_eq!(candidates[0].id, "open");
789 assert_eq!(candidates[1].id, "completed");
790 }
791
792 #[test]
793 fn explicit_request_requeues_a_permanent_failure() {
794 let (db, _dir) = crate::db::test_helpers::test_db();
795 let mut issue = make_issue("CUT-1", "CUT");
796 issue.id = "issue-1".into();
797 db.upsert_issue(&issue).unwrap();
798 db.ensure_hydration_state_for_issue("default", &issue, "initial")
799 .unwrap();
800 db.mark_hydration_failed(
801 "default",
802 "issue-1",
803 HydrationResource::Comments,
804 HydrationStatus::PermissionDenied,
805 None,
806 "forbidden",
807 )
808 .unwrap();
809
810 db.requeue_issue_hydration("default", "issue-1", "explicit")
811 .unwrap();
812 let state = db.get_issue_hydration_state("default", "issue-1").unwrap();
813 assert!(state
814 .resources
815 .iter()
816 .all(|resource| resource.status == HydrationStatus::Pending));
817 }
818
819 #[test]
820 fn retry_state_survives_reopening_database() {
821 let dir = tempfile::tempdir().unwrap();
822 let path = dir.path().join("retry.db");
823 {
824 let db = Database::open(&path).unwrap();
825 let mut issue = make_issue("CUT-1", "CUT");
826 issue.id = "issue-1".into();
827 db.upsert_issue(&issue).unwrap();
828 db.ensure_hydration_state_for_issue("default", &issue, "initial")
829 .unwrap();
830 db.mark_hydration_running(
831 "default",
832 "issue-1",
833 HydrationResource::Relations,
834 "2026-01-01T00:00:00Z",
835 )
836 .unwrap();
837 }
838 let reopened = Database::open(&path).unwrap();
839 let state = reopened
840 .get_issue_hydration_state("default", "issue-1")
841 .unwrap();
842 let relations = state
843 .resources
844 .iter()
845 .find(|resource| resource.resource == HydrationResource::Relations)
846 .unwrap();
847 assert_eq!(relations.status, HydrationStatus::Retryable);
848 assert_eq!(relations.attempt_count, 1);
849 assert!(relations.next_retry_at.is_some());
850 }
851
852 #[test]
853 fn migration_12_preserves_issues_comments_and_comment_sync_evidence() {
854 let dir = tempfile::tempdir().unwrap();
855 let path = dir.path().join("migration.db");
856 {
857 let db = Database::open(&path).unwrap();
858 let mut issue = make_issue("CUT-1", "CUT");
859 issue.id = "issue-1".into();
860 db.upsert_issue(&issue).unwrap();
861 db.replace_issue_comments(
862 "issue-1",
863 "default",
864 &[crate::db::Comment {
865 id: "comment-1".into(),
866 issue_id: "issue-1".into(),
867 body: "preserved".into(),
868 user_name: None,
869 created_at: "2026-01-01T00:00:00Z".into(),
870 updated_at: None,
871 parent_id: None,
872 url: None,
873 workspace_id: "default".into(),
874 }],
875 )
876 .unwrap();
877 db.mark_comments_synced("issue-1", "default", 1).unwrap();
878 db.set_sync_cursor("default", "CUT", "2026-01-02T00:00:00Z")
879 .unwrap();
880 db.with_conn(|conn| {
881 conn.execute_batch(
882 "DROP TABLE issue_hydration_state;
883 DELETE FROM schema_version WHERE version = 12;",
884 )?;
885 Ok(())
886 })
887 .unwrap();
888 }
889
890 let migrated = Database::open(&path).unwrap();
891 assert!(migrated.get_issue("CUT-1").unwrap().is_some());
892 assert_eq!(migrated.get_comments("issue-1").unwrap().len(), 1);
893 assert_eq!(
894 migrated
895 .get_issue_hydration_state("default", "issue-1")
896 .unwrap()
897 .resources
898 .iter()
899 .find(|resource| resource.resource == HydrationResource::Comments)
900 .unwrap()
901 .status,
902 HydrationStatus::Hydrated
903 );
904 assert_eq!(
905 migrated
906 .get_synced_through_at("default", "CUT")
907 .unwrap()
908 .as_deref(),
909 Some("2026-01-02T00:00:00Z")
910 );
911 }
912}