feat(backend): ingest GitHub issues with idempotent upsert (#74)

backend/Cargo.toml

@@ -21,6 +21,7 @@ sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "sqlit
 async-trait = "0.1"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
+reqwest = { version = "0.11", features = ["json", "rustls-tls"] }
 
 # Authentication & Crypto
 siwe = "0.6"

backend/migrations/003_github_issues.sql

@@ -0,0 +1,23 @@
+-- GitHub Issues ingestion table
+CREATE TABLE IF NOT EXISTS github_issues (
+    repo TEXT NOT NULL,
+    repo_id BIGINT NOT NULL,
+    github_issue_id BIGINT NOT NULL,
+    number INT NOT NULL,
+    title TEXT NOT NULL,
+    state TEXT NOT NULL CHECK (state IN ('open','closed')),
+    labels JSONB NOT NULL DEFAULT '[]'::jsonb,
+    points INT NOT NULL DEFAULT 0,
+    assignee_logins JSONB NOT NULL DEFAULT '[]'::jsonb,
+    html_url TEXT NOT NULL,
+    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    closed_at TIMESTAMP WITH TIME ZONE NULL,
+    rewarded BOOL NOT NULL DEFAULT FALSE,
+    distribution_id TEXT NULL,
+    updated_at TIMESTAMP WITH TIME ZONE NOT NULL,
+    PRIMARY KEY (repo_id, github_issue_id)
+);
+
+CREATE INDEX IF NOT EXISTS idx_github_issues_repo ON github_issues(repo);
+CREATE INDEX IF NOT EXISTS idx_github_issues_state ON github_issues(state);
+CREATE INDEX IF NOT EXISTS idx_github_issues_number ON github_issues(number);

backend/src/application/commands/github_sync.rs

@@ -0,0 +1,72 @@
+use anyhow::Result;
+// chrono imports not needed in this module after refactor
+
+use crate::domain::{
+    entities::github_issue::GithubIssue,
+    repositories::GithubIssueRepository,
+    services::github_api_service::{GithubApiService, GithubIssueApi, GithubLabel, GithubRepoApi},
+};
+
+pub fn derive_points(labels: &[GithubLabel]) -> i32 {
+    for l in labels {
+        let name = l.name.to_lowercase();
+        if let Some(rest) = name.strip_prefix("points:") {
+            if let Ok(v) = rest.trim().parse::<i32>() {
+                return v.max(0);
+            }
+        }
+    }
+    0
+}
+
+pub fn transform_issue(repo_api: &GithubRepoApi, ia: &GithubIssueApi) -> Option<GithubIssue> {
+    // Ignore PRs
+    if ia.pull_request.is_some() {
+        return None;
+    }
+    let points = derive_points(&ia.labels);
+
+    let label_names: Vec<String> = ia.labels.iter().map(|l| l.name.to_lowercase()).collect();
+    let assignee_logins: Vec<String> = ia.assignees.iter().map(|a| a.login.clone()).collect();
+
+    Some(GithubIssue {
+        repo: repo_api.full_name.clone(),
+        repo_id: repo_api.id,
+        github_issue_id: ia.id,
+        number: ia.number,
+        title: ia.title.clone(),
+        state: ia.state.clone(),
+        labels: serde_json::Value::from(label_names),
+        points,
+        assignee_logins: serde_json::Value::from(assignee_logins),
+        html_url: ia.html_url.clone(),
+        created_at: ia.created_at,
+        closed_at: ia.closed_at,
+        rewarded: false,
+        distribution_id: None,
+        updated_at: ia.updated_at,
+    })
+}
+
+pub async fn sync_github_issues<R: GithubIssueRepository + ?Sized, A: GithubApiService + ?Sized>(
+    repo: &R,
+    api: &A,
+    repos: &[String],
+    since: Option<String>,
+) -> Result<()> {
+    let mut all_issues: Vec<GithubIssue> = Vec::new();
+
+    for repo_full in repos {
+        let repo_api = api.get_repo(repo_full).await?;
+        let issues_api = api.list_issues(repo_full, since.as_deref()).await?;
+
+        for ia in issues_api.iter() {
+            if let Some(issue) = transform_issue(&repo_api, ia) {
+                all_issues.push(issue);
+            }
+        }
+    }
+
+    repo.upsert_issues(&all_issues).await?;
+    Ok(())
+}

backend/src/application/commands/mod.rs

@@ -1,4 +1,5 @@
 pub mod create_profile;
 pub mod get_all_profiles;
 pub mod get_profile;
+pub mod github_sync;
 pub mod update_profile;

backend/src/domain/entities/github_issue.rs

@@ -0,0 +1,21 @@
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct GithubIssue {
+    pub repo: String,
+    pub repo_id: i64,
+    pub github_issue_id: i64,
+    pub number: i32,
+    pub title: String,
+    pub state: String,             // 'open' | 'closed'
+    pub labels: serde_json::Value, // JSON array of label names
+    pub points: i32,
+    pub assignee_logins: serde_json::Value, // JSON array of logins
+    pub html_url: String,
+    pub created_at: DateTime<Utc>,
+    pub closed_at: Option<DateTime<Utc>>,
+    pub rewarded: bool,
+    pub distribution_id: Option<String>,
+    pub updated_at: DateTime<Utc>,
+}

backend/src/domain/entities/mod.rs

@@ -1,3 +1,4 @@
+pub mod github_issue;
 pub mod profile;
 
 pub use profile::Profile;

backend/src/domain/repositories/github_issue_repository.rs

@@ -0,0 +1,8 @@
+use async_trait::async_trait;
+
+use crate::domain::entities::github_issue::GithubIssue;
+
+#[async_trait]
+pub trait GithubIssueRepository: Send + Sync {
+    async fn upsert_issues(&self, issues: &[GithubIssue]) -> anyhow::Result<()>;
+}

backend/src/domain/repositories/mod.rs

@@ -1,3 +1,5 @@
+pub mod github_issue_repository;
 pub mod profile_repository;
 
+pub use github_issue_repository::GithubIssueRepository;
 pub use profile_repository::ProfileRepository;

backend/src/domain/services/github_api_service.rs

@@ -0,0 +1,44 @@
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use serde::Deserialize;
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct GithubRepoApi {
+    pub id: i64,
+    pub full_name: String, // org/repo
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct GithubLabel {
+    pub name: String,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct GithubAssignee {
+    pub login: String,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct GithubIssueApi {
+    pub id: i64,
+    pub number: i32,
+    pub title: String,
+    pub state: String,
+    pub html_url: String,
+    pub pull_request: Option<serde_json::Value>,
+    pub labels: Vec<GithubLabel>,
+    pub assignees: Vec<GithubAssignee>,
+    pub created_at: DateTime<Utc>,
+    pub closed_at: Option<DateTime<Utc>>,
+    pub updated_at: DateTime<Utc>,
+}
+
+#[async_trait]
+pub trait GithubApiService: Send + Sync {
+    async fn get_repo(&self, repo_full: &str) -> anyhow::Result<GithubRepoApi>;
+    async fn list_issues(
+        &self,
+        repo_full: &str,
+        since: Option<&str>,
+    ) -> anyhow::Result<Vec<GithubIssueApi>>;
+}

backend/src/domain/services/mod.rs

@@ -1 +1,2 @@
 pub mod auth_service;
+pub mod github_api_service;

backend/src/infrastructure/repositories/mod.rs

@@ -1,3 +1,5 @@
+pub mod postgres_github_issue_repository;
 pub mod postgres_profile_repository;
 
+pub use postgres_github_issue_repository::PostgresGithubIssueRepository;
 pub use postgres_profile_repository::PostgresProfileRepository;

backend/src/infrastructure/repositories/postgres_github_issue_repository.rs

@@ -0,0 +1,68 @@
+use async_trait::async_trait;
+use sqlx::PgPool;
+
+use crate::domain::entities::github_issue::GithubIssue;
+use crate::domain::repositories::github_issue_repository::GithubIssueRepository;
+
+#[derive(Clone)]
+pub struct PostgresGithubIssueRepository {
+    pool: PgPool,
+}
+
+impl PostgresGithubIssueRepository {
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+#[async_trait]
+impl GithubIssueRepository for PostgresGithubIssueRepository {
+    async fn upsert_issues(&self, issues: &[GithubIssue]) -> anyhow::Result<()> {
+        let mut tx = self.pool.begin().await?;
+        for issue in issues {
+            sqlx::query!(
+                r#"
+                INSERT INTO github_issues (
+                    repo, repo_id, github_issue_id, number, title, state, labels, points,
+                    assignee_logins, html_url, created_at, closed_at, rewarded, distribution_id, updated_at
+                ) VALUES (
+                    $1, $2, $3, $4, $5, $6, $7, $8,
+                    $9, $10, $11, $12, $13, $14, $15
+                )
+                ON CONFLICT (repo_id, github_issue_id) DO UPDATE SET
+                    repo = EXCLUDED.repo,
+                    number = EXCLUDED.number,
+                    title = EXCLUDED.title,
+                    state = EXCLUDED.state,
+                    labels = EXCLUDED.labels,
+                    points = EXCLUDED.points,
+                    assignee_logins = EXCLUDED.assignee_logins,
+                    html_url = EXCLUDED.html_url,
+                    created_at = EXCLUDED.created_at,
+                    closed_at = EXCLUDED.closed_at,
+                    rewarded = github_issues.rewarded,
+                    distribution_id = COALESCE(EXCLUDED.distribution_id, github_issues.distribution_id),
+                    updated_at = EXCLUDED.updated_at
+                "#,
+                issue.repo,
+                issue.repo_id,
+                issue.github_issue_id,
+                issue.number,
+                issue.title,
+                issue.state,
+                issue.labels,
+                issue.points,
+                issue.assignee_logins,
+                issue.html_url,
+                issue.created_at,
+                issue.closed_at,
+                issue.rewarded,
+                issue.distribution_id,
+                issue.updated_at
+            )
+            .execute(&mut *tx)
+            .await?;
+        }
+        tx.commit().await?;
+        Ok(())
+    }
+}

backend/src/infrastructure/services/github_api_http_service.rs

@@ -0,0 +1,64 @@
+use anyhow::Result;
+use async_trait::async_trait;
+
+use crate::domain::services::github_api_service::{
+    GithubApiService, GithubIssueApi, GithubRepoApi,
+};
+
+#[derive(Clone)]
+pub struct GithubApiHttpService {
+    client: reqwest::Client,
+}
+
+impl GithubApiHttpService {
+    pub fn new() -> Self {
+        Self {
+            client: reqwest::Client::new(),
+        }
+    }
+}
+impl Default for GithubApiHttpService {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl GithubApiService for GithubApiHttpService {
+    async fn get_repo(&self, repo_full: &str) -> Result<GithubRepoApi> {
+        let repo_api: GithubRepoApi = self
+            .client
+            .get(format!("https://api.github.com/repos/{}", repo_full))
+            .header("User-Agent", "guild-backend")
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await?;
+        Ok(repo_api)
+    }
+
+    async fn list_issues(
+        &self,
+        repo_full: &str,
+        since: Option<&str>,
+    ) -> Result<Vec<GithubIssueApi>> {
+        let mut url = format!(
+            "https://api.github.com/repos/{}/issues?state=all",
+            repo_full,
+        );
+        if let Some(s) = since {
+            url.push_str(&format!("&since={}", s));
+        }
+        let issues_api: Vec<GithubIssueApi> = self
+            .client
+            .get(url)
+            .header("User-Agent", "guild-backend")
+            .send()
+            .await?
+            .error_for_status()?
+            .json()
+            .await?;
+        Ok(issues_api)
+    }
+}

backend/src/infrastructure/services/mod.rs

@@ -1 +1,2 @@
 pub mod ethereum_address_verification_service;
+pub mod github_api_http_service;

backend/src/lib.rs

@@ -1,5 +1,4 @@
 pub mod application;
-pub mod database;
 pub mod domain;
 pub mod infrastructure;
 pub mod presentation;