Merge pull request #14 from vectordotdev/feat/fetch-since-and-compact

Add --since to fetch commands and compact for deduplication

Author: pront

src/commands/compact.rs

@@ -0,0 +1,87 @@
+use anyhow::{Context, Result};
+use serde_json::Value;
+use std::collections::HashMap;
+use std::fs;
+use std::path::Path;
+
+/// Deduplicate JSON year files in a directory.
+/// For issues/PRs: dedup by "id". For discussions: dedup by "number".
+pub fn run(dir: &str) -> Result<()> {
+    let path = Path::new(dir);
+    if !path.is_dir() {
+        anyhow::bail!("{dir} is not a directory");
+    }
+
+    let mut entries: Vec<_> = fs::read_dir(path)?
+        .filter_map(|e| e.ok())
+        .filter(|e| e.path().extension().is_some_and(|ext| ext == "json"))
+        .collect();
+    entries.sort_by_key(|e| e.path());
+
+    if entries.is_empty() {
+        println!("No JSON files found in {dir}");
+        return Ok(());
+    }
+
+    // Detect key field from first file
+    let first_json = fs::read_to_string(entries[0].path())?;
+    let first_items: Vec<Value> = serde_json::from_str(&first_json)?;
+    let key_field = if first_items.first().and_then(|v| v.get("number")).is_some()
+        && first_items.first().and_then(|v| v.get("id")).is_none()
+    {
+        "number"
+    } else {
+        "id"
+    };
+    println!("Deduplicating by \"{key_field}\" in {dir}");
+
+    let mut total_before = 0;
+    let mut total_after = 0;
+
+    for entry in entries {
+        let file_path = entry.path();
+        let json = fs::read_to_string(&file_path)
+            .with_context(|| format!("Failed to read {}", file_path.display()))?;
+        let items: Vec<Value> = serde_json::from_str(&json)?;
+        let before = items.len();
+
+        // Dedup: last occurrence wins (preserves most recent state)
+        let mut seen: HashMap<String, usize> = HashMap::new();
+        let mut deduped: Vec<Value> = Vec::new();
+        for item in items {
+            let key = item[key_field].to_string();
+            if let Some(idx) = seen.get(&key) {
+                deduped[*idx] = item;
+            } else {
+                seen.insert(key, deduped.len());
+                deduped.push(item);
+            }
+        }
+
+        let after = deduped.len();
+        total_before += before;
+        total_after += after;
+
+        if before != after {
+            let json_str = serde_json::to_string_pretty(&deduped)?;
+            fs::write(&file_path, json_str)?;
+            println!(
+                "  {}: {} -> {} (removed {} duplicates)",
+                file_path.display(),
+                before,
+                after,
+                before - after
+            );
+        } else {
+            println!("  {}: {} items (no duplicates)", file_path.display(), after);
+        }
+    }
+
+    println!(
+        "Total: {} -> {} (removed {} duplicates)",
+        total_before,
+        total_after,
+        total_before - total_after
+    );
+    Ok(())
+}

src/commands/fetch_discussions.rs

@@ -1,8 +1,10 @@
+use crate::commands::fetch_issues::parse_since;
 use crate::config::Config;
 use anyhow::{Context, Result};
 use reqwest::blocking::Client;
 use serde::{Deserialize, Serialize};
-use serde_json::json;
+use serde_json::{json, Value};
+use std::collections::BTreeMap;
 use std::fs;
 use std::path::Path;
 
@@ -45,6 +47,42 @@ query($owner: String!, $name: String!, $first: Int!, $after: String) {
 }
 "#;
 
+const QUERY_SINCE: &str = r#"
+query($owner: String!, $name: String!, $first: Int!, $after: String) {
+  repository(owner: $owner, name: $name) {
+    discussions(first: $first, after: $after, orderBy: {field: UPDATED_AT, direction: DESC}) {
+      pageInfo {
+        endCursor
+        hasNextPage
+      }
+      nodes {
+        number
+        title
+        bodyText
+        url
+        createdAt
+        updatedAt
+        closedAt
+        closed
+        stateReason
+        isAnswered
+        locked
+        author {
+          login
+        }
+        category {
+          name
+        }
+        comments {
+          totalCount
+        }
+        upvoteCount
+      }
+    }
+  }
+}
+"#;
+
 #[derive(Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct Discussion {
@@ -111,28 +149,129 @@ struct GraphQlResponse {
     errors: Option<serde_json::Value>,
 }
 
-pub fn run(config: &Config) -> Result<()> {
-    run_with_client(&Client::new(), config)
+pub fn run(config: &Config, since: Option<&str>) -> Result<()> {
+    run_with_client(&Client::new(), config, since)
 }
 
-pub fn run_with_client(client: &Client, config: &Config) -> Result<()> {
-    let discussions = fetch_all_discussions(client, config)?;
+pub fn run_with_client(client: &Client, config: &Config, since: Option<&str>) -> Result<()> {
+    let since_ts = since.map(parse_since).transpose()?;
+
+    let discussions = if let Some(ref ts) = since_ts {
+        println!("Fetching discussions updated since {ts}...");
+        fetch_discussions_since(client, config, ts)?
+    } else {
+        fetch_all_discussions(client, config)?
+    };
 
     println!("Total discussions fetched: {}", discussions.len());
 
-    let out_dir = Path::new("out/historical/discussions");
-    fs::create_dir_all(out_dir)?;
-    let out_file = out_dir.join(format!(
-        "{}_{}_discussions.json",
-        config.repo_owner, config.repo_name
-    ));
-    let json = serde_json::to_string_pretty(&discussions)?;
-    fs::write(&out_file, json)?;
-    println!("Saved to '{}'", out_file.display());
+    let repo_prefix = format!("{}_{}", config.repo_owner, config.repo_name);
+    let out_dir = Path::new("data").join(&repo_prefix).join("discussions");
+    fs::create_dir_all(&out_dir)?;
+
+    // Bucket by year based on createdAt
+    let mut by_year: BTreeMap<String, Vec<Discussion>> = BTreeMap::new();
+    for d in discussions {
+        let year = d.created_at.get(..4).unwrap_or("unknown").to_string();
+        by_year.entry(year).or_default().push(d);
+    }
+
+    for (year, items) in &by_year {
+        let path = out_dir.join(format!("{year}.json"));
+        let mut existing: Vec<Value> = if path.exists() {
+            let json = fs::read_to_string(&path)?;
+            serde_json::from_str(&json).unwrap_or_default()
+        } else {
+            Vec::new()
+        };
+        let new_values: Vec<Value> = items.iter()
+            .map(|d| serde_json::to_value(d).unwrap())
+            .collect();
+        existing.extend(new_values);
+        let json_str = serde_json::to_string_pretty(&existing)?;
+        fs::write(&path, json_str)?;
+        println!("  {year}.json: appended {} discussions", items.len());
+    }
 
     Ok(())
 }
 
+fn fetch_discussions_since(client: &Client, config: &Config, since: &str) -> Result<Vec<Discussion>> {
+    let mut discussions = Vec::new();
+    let mut after: Option<String> = None;
+    let mut page = 1;
+
+    loop {
+        println!("Fetching page {page} of discussions...");
+
+        let body = json!({
+            "query": QUERY_SINCE,
+            "variables": {
+                "owner": config.repo_owner,
+                "name": config.repo_name,
+                "first": PAGE_SIZE,
+                "after": after,
+            }
+        });
+
+        let response = client
+            .post(GRAPHQL_URL)
+            .header("Authorization", format!("Bearer {}", config.github_token))
+            .header("Accept", "application/vnd.github+json")
+            .header("User-Agent", "github-tools")
+            .json(&body)
+            .send()
+            .context("Failed to send GraphQL request")?;
+
+        if !response.status().is_success() {
+            let status = response.status();
+            let text = response.text().unwrap_or_default();
+            eprintln!("Warning: GraphQL request failed ({status}): {text}");
+            break;
+        }
+
+        let result: GraphQlResponse = response
+            .json()
+            .context("Failed to parse GraphQL response")?;
+
+        if let Some(errors) = result.errors {
+            eprintln!("Warning: GraphQL errors: {errors}");
+            break;
+        }
+
+        let page_data = result
+            .data
+            .context("Missing 'data' in GraphQL response")?
+            .repository
+            .discussions;
+
+        let has_next = page_data.page_info.has_next_page;
+        after = page_data.page_info.end_cursor;
+
+        let mut hit_boundary = false;
+        for d in page_data.nodes {
+            if d.updated_at.as_str() < since {
+                hit_boundary = true;
+            } else {
+                discussions.push(d);
+            }
+        }
+
+        println!("Fetched {} discussions so far...", discussions.len());
+
+        if hit_boundary {
+            println!("Reached discussions older than --since cutoff.");
+            break;
+        }
+        if !has_next {
+            break;
+        }
+        page += 1;
+    }
+
+    Ok(discussions)
+}
+
 fn fetch_all_discussions(client: &Client, config: &Config) -> Result<Vec<Discussion>> {
     let mut discussions = Vec::new();
     let mut after: Option<String> = None;