use axum::{extract::State, http::HeaderMap, response::Json};
use super::state::MultiUserMemoryManager;
use crate::errors::{AppError, ValidationErrorExt};
use crate::integrations;
use crate::memory::{self, Experience, ExperienceType};
use crate::validation;
use std::sync::Arc;
type AppState = Arc<MultiUserMemoryManager>;
#[tracing::instrument(skip(state, body, headers))]
pub async fn linear_webhook(
State(state): State<AppState>,
headers: HeaderMap,
body: axum::body::Bytes,
) -> Result<Json<serde_json::Value>, AppError> {
use integrations::linear::LinearWebhook;
let signing_secret = std::env::var("LINEAR_WEBHOOK_SECRET").ok();
let has_secret = signing_secret.is_some();
let webhook = LinearWebhook::new(signing_secret);
let signature = headers
.get("linear-signature")
.and_then(|h| h.to_str().ok());
match (has_secret, signature) {
(true, Some(sig)) => {
if !webhook
.verify_signature(&body, sig)
.map_err(AppError::Internal)?
{
return Err(AppError::InvalidInput {
field: "signature".to_string(),
reason: "Invalid webhook signature".to_string(),
});
}
}
(true, None) => {
return Err(AppError::InvalidInput {
field: "linear-signature".to_string(),
reason: "Missing required webhook signature header".to_string(),
});
}
(false, _) => {
tracing::warn!("No LINEAR_WEBHOOK_SECRET configured, skipping signature verification");
}
}
let payload = webhook.parse_payload(&body).map_err(AppError::Internal)?;
if payload.entity_type != "Issue" {
return Ok(Json(serde_json::json!({
"status": "ignored",
"reason": "Only Issue events are processed"
})));
}
if payload.action == "remove" {
return Ok(Json(serde_json::json!({
"status": "acknowledged",
"action": "remove"
})));
}
let external_id = match &payload.data.identifier {
Some(id) => format!("linear:{}", id),
None => format!("linear:{}", payload.data.id),
};
let content = LinearWebhook::issue_to_content(&payload.data);
let tags = LinearWebhook::issue_to_tags(&payload.data);
let change_type = LinearWebhook::determine_change_type(&payload.action, &payload.data);
let user_id =
std::env::var("LINEAR_SYNC_USER_ID").unwrap_or_else(|_| "linear-sync".to_string());
let experience = Experience {
content: content.clone(),
experience_type: ExperienceType::Task,
entities: tags.clone(),
..Default::default()
};
let change_type_enum = match change_type.as_str() {
"created" => memory::types::ChangeType::Created,
"status_changed" => memory::types::ChangeType::StatusChanged,
"tags_updated" => memory::types::ChangeType::TagsUpdated,
_ => memory::types::ChangeType::ContentUpdated,
};
let memory_system = state
.get_user_memory(&user_id)
.map_err(AppError::Internal)?;
let (memory_id, was_update) = {
let memory = memory_system.clone();
let ext_id = external_id.clone();
let exp = experience.clone();
let ct = change_type_enum;
let actor_name = payload
.actor
.as_ref()
.and_then(|a| a.name.clone())
.unwrap_or_else(|| "linear-webhook".to_string());
tokio::task::spawn_blocking(move || {
let memory_guard = memory.read();
memory_guard.upsert(ext_id, exp, ct, Some(actor_name), None)
})
.await
.map_err(|e| AppError::Internal(anyhow::anyhow!("Blocking task panicked: {e}")))?
.map_err(AppError::Internal)?
};
Ok(Json(serde_json::json!({
"status": "success",
"id": memory_id.0.to_string(),
"external_id": external_id,
"was_update": was_update,
"action": payload.action
})))
}
#[tracing::instrument(skip(state, req), fields(user_id = %req.user_id))]
pub async fn linear_sync(
State(state): State<AppState>,
Json(req): Json<integrations::linear::LinearSyncRequest>,
) -> Result<Json<integrations::linear::LinearSyncResponse>, AppError> {
use integrations::linear::{LinearClient, LinearSyncResponse, LinearWebhook};
validation::validate_user_id(&req.user_id).map_validation_err("user_id")?;
if req.api_key.is_empty() {
return Err(AppError::InvalidInput {
field: "api_key".to_string(),
reason: "Linear API key is required".to_string(),
});
}
let client = LinearClient::new(req.api_key.clone());
let issues = client
.fetch_issues(
req.team_id.as_deref(),
req.updated_after.as_deref(),
req.limit,
)
.await
.map_err(|e| AppError::Internal(anyhow::anyhow!("Failed to fetch Linear issues: {}", e)))?;
let total = issues.len();
let mut created_count = 0;
let mut updated_count = 0;
let mut error_count = 0;
let mut errors = Vec::new();
let memory_system = state
.get_user_memory(&req.user_id)
.map_err(AppError::Internal)?;
for issue in issues {
let external_id = match &issue.identifier {
Some(id) => format!("linear:{}", id),
None => format!("linear:{}", issue.id),
};
let content = LinearWebhook::issue_to_content(&issue);
let tags = LinearWebhook::issue_to_tags(&issue);
let experience = Experience {
content,
experience_type: ExperienceType::Task,
entities: tags,
..Default::default()
};
let result = {
let memory = memory_system.clone();
let ext_id = external_id.clone();
let exp = experience;
tokio::task::spawn_blocking(move || {
let memory_guard = memory.read();
memory_guard.upsert(
ext_id,
exp,
memory::types::ChangeType::ContentUpdated,
Some("linear-bulk-sync".to_string()),
None,
)
})
.await
};
match result {
Ok(Ok((_, was_update))) => {
if was_update {
updated_count += 1;
} else {
created_count += 1;
}
}
Ok(Err(e)) => {
error_count += 1;
errors.push(format!("{}: {}", external_id, e));
}
Err(e) => {
error_count += 1;
errors.push(format!("{}: Task panicked: {}", external_id, e));
}
}
}
Ok(Json(LinearSyncResponse {
synced_count: total,
created_count,
updated_count,
error_count,
errors,
}))
}
#[tracing::instrument(skip(state, body, headers))]
pub async fn github_webhook(
State(state): State<AppState>,
headers: HeaderMap,
body: axum::body::Bytes,
) -> Result<Json<serde_json::Value>, AppError> {
use integrations::github::GitHubWebhook;
let webhook_secret = std::env::var("GITHUB_WEBHOOK_SECRET").ok();
let has_secret = webhook_secret.is_some();
let webhook = GitHubWebhook::new(webhook_secret);
let signature = headers
.get("x-hub-signature-256")
.and_then(|h| h.to_str().ok());
match (has_secret, signature) {
(true, Some(sig)) => {
if !webhook
.verify_signature(&body, sig)
.map_err(AppError::Internal)?
{
return Err(AppError::InvalidInput {
field: "signature".to_string(),
reason: "Invalid webhook signature".to_string(),
});
}
}
(true, None) => {
return Err(AppError::InvalidInput {
field: "x-hub-signature-256".to_string(),
reason: "Missing required webhook signature header".to_string(),
});
}
(false, _) => {
tracing::warn!("No GITHUB_WEBHOOK_SECRET configured, skipping signature verification");
}
}
let event_type = headers
.get("x-github-event")
.and_then(|h| h.to_str().ok())
.unwrap_or("unknown");
if event_type != "issues" && event_type != "pull_request" {
return Ok(Json(serde_json::json!({
"status": "ignored",
"reason": format!("Only issues and pull_request events are processed, got: {}", event_type)
})));
}
let payload = webhook.parse_payload(&body).map_err(AppError::Internal)?;
let user_id =
std::env::var("GITHUB_SYNC_USER_ID").unwrap_or_else(|_| "github-sync".to_string());
let (external_id, content, tags, change_type) = if let Some(issue) = &payload.issue {
let ext_id = GitHubWebhook::issue_external_id(&payload.repository, issue.number);
let content = GitHubWebhook::issue_to_content(issue, &payload.repository);
let tags = GitHubWebhook::issue_to_tags(issue, &payload.repository);
let ct = GitHubWebhook::determine_change_type(&payload.action, false);
(ext_id, content, tags, ct)
} else if let Some(pr) = &payload.pull_request {
let ext_id = GitHubWebhook::pr_external_id(&payload.repository, pr.number);
let content = GitHubWebhook::pr_to_content(pr, &payload.repository);
let tags = GitHubWebhook::pr_to_tags(pr, &payload.repository);
let ct = GitHubWebhook::determine_change_type(&payload.action, true);
(ext_id, content, tags, ct)
} else {
return Ok(Json(serde_json::json!({
"status": "ignored",
"reason": "No issue or pull_request data in payload"
})));
};
let experience = Experience {
content: content.clone(),
experience_type: ExperienceType::Task,
entities: tags.clone(),
..Default::default()
};
let change_type_enum = match change_type.as_str() {
"created" => memory::types::ChangeType::Created,
"status_changed" => memory::types::ChangeType::StatusChanged,
"tags_updated" => memory::types::ChangeType::TagsUpdated,
_ => memory::types::ChangeType::ContentUpdated,
};
let memory_system = state
.get_user_memory(&user_id)
.map_err(AppError::Internal)?;
let (memory_id, was_update) = {
let memory = memory_system.clone();
let ext_id = external_id.clone();
let exp = experience.clone();
let ct = change_type_enum;
let actor_name = payload
.sender
.as_ref()
.map(|s| s.login.clone())
.unwrap_or_else(|| "github-webhook".to_string());
tokio::task::spawn_blocking(move || {
let memory_guard = memory.read();
memory_guard.upsert(ext_id, exp, ct, Some(actor_name), None)
})
.await
.map_err(|e| AppError::Internal(anyhow::anyhow!("Blocking task panicked: {e}")))?
.map_err(AppError::Internal)?
};
Ok(Json(serde_json::json!({
"status": "success",
"id": memory_id.0.to_string(),
"external_id": external_id,
"was_update": was_update,
"action": payload.action,
"event_type": event_type
})))
}
#[tracing::instrument(skip(state, req), fields(user_id = %req.user_id))]
pub async fn github_sync(
State(state): State<AppState>,
Json(req): Json<integrations::github::GitHubSyncRequest>,
) -> Result<Json<integrations::github::GitHubSyncResponse>, AppError> {
use integrations::github::{GitHubClient, GitHubSyncResponse, GitHubWebhook};
validation::validate_user_id(&req.user_id).map_validation_err("user_id")?;
if req.token.is_empty() {
return Err(AppError::InvalidInput {
field: "token".to_string(),
reason: "GitHub token is required".to_string(),
});
}
let client = GitHubClient::new(req.token.clone());
let repo_info = client
.get_repository(&req.owner, &req.repo)
.await
.map_err(|e| AppError::Internal(anyhow::anyhow!("Failed to get repository: {}", e)))?;
let mut issues_synced = 0;
let mut prs_synced = 0;
let mut created_count = 0;
let mut updated_count = 0;
let mut error_count = 0;
let mut errors = Vec::new();
let memory_system = state
.get_user_memory(&req.user_id)
.map_err(AppError::Internal)?;
if req.sync_issues {
let issues = client
.fetch_issues(&req.owner, &req.repo, &req.state, req.limit)
.await
.map_err(|e| AppError::Internal(anyhow::anyhow!("Failed to fetch issues: {}", e)))?;
for issue in issues {
let external_id = GitHubWebhook::issue_external_id(&repo_info, issue.number);
let content = GitHubWebhook::issue_to_content(&issue, &repo_info);
let tags = GitHubWebhook::issue_to_tags(&issue, &repo_info);
let experience = Experience {
content,
experience_type: ExperienceType::Task,
entities: tags,
..Default::default()
};
let result = {
let memory = memory_system.clone();
let ext_id = external_id.clone();
let exp = experience;
tokio::task::spawn_blocking(move || {
let memory_guard = memory.read();
memory_guard.upsert(
ext_id,
exp,
memory::types::ChangeType::ContentUpdated,
Some("github-bulk-sync".to_string()),
None,
)
})
.await
};
match result {
Ok(Ok((_, was_update))) => {
issues_synced += 1;
if was_update {
updated_count += 1;
} else {
created_count += 1;
}
}
Ok(Err(e)) => {
error_count += 1;
errors.push(format!("{}: {}", external_id, e));
}
Err(e) => {
error_count += 1;
errors.push(format!("{}: {}", external_id, e));
}
}
}
}
if req.sync_prs {
let prs = client
.fetch_pull_requests(&req.owner, &req.repo, &req.state, req.limit)
.await
.map_err(|e| AppError::Internal(anyhow::anyhow!("Failed to fetch PRs: {}", e)))?;
for pr in prs {
let external_id = GitHubWebhook::pr_external_id(&repo_info, pr.number);
let content = GitHubWebhook::pr_to_content(&pr, &repo_info);
let tags = GitHubWebhook::pr_to_tags(&pr, &repo_info);
let experience = Experience {
content,
experience_type: ExperienceType::Task,
entities: tags,
..Default::default()
};
let result = {
let memory = memory_system.clone();
let ext_id = external_id.clone();
let exp = experience;
tokio::task::spawn_blocking(move || {
let memory_guard = memory.read();
memory_guard.upsert(
ext_id,
exp,
memory::types::ChangeType::ContentUpdated,
Some("github-bulk-sync".to_string()),
None,
)
})
.await
};
match result {
Ok(Ok((_, was_update))) => {
prs_synced += 1;
if was_update {
updated_count += 1;
} else {
created_count += 1;
}
}
Ok(Err(e)) => {
error_count += 1;
errors.push(format!("{}: {}", external_id, e));
}
Err(e) => {
error_count += 1;
errors.push(format!("{}: {}", external_id, e));
}
}
}
}
Ok(Json(GitHubSyncResponse {
synced_count: issues_synced + prs_synced,
issues_synced,
prs_synced,
commits_synced: 0,
created_count,
updated_count,
error_count,
errors,
}))
}