feat(sync): OneDrive (Graph) provider + kind-based target dispatch + token resolution (T9.10)
This commit is contained in:
+177
-4
@@ -15,6 +15,9 @@ use crate::models::{MeetingId, SyncJobInfo, SyncKind};
|
||||
use async_trait::async_trait;
|
||||
use std::path::Path;
|
||||
|
||||
#[cfg(feature = "sync")]
|
||||
pub mod oauth;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum SyncError {
|
||||
#[error("target unreachable: {0}")]
|
||||
@@ -436,10 +439,180 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
// Secondary OAuth providers (Phase 9b) — same trait, separate impls.
|
||||
#[cfg(feature = "sync")]
|
||||
pub struct OneDriveTarget; // MS Graph
|
||||
fn now_unix() -> i64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_secs() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Resolve a live access token for an OAuth target, refreshing (and re-storing)
|
||||
/// it if it's expired. Tokens live in the OS credential store as JSON.
|
||||
#[cfg(feature = "sync")]
|
||||
pub struct DropboxTarget;
|
||||
async fn resolve_access_token(kind: &str, credential_ref: &str) -> Result<String, SyncError> {
|
||||
let provider = oauth::provider_for(kind)
|
||||
.ok_or_else(|| SyncError::Upload(format!("unknown oauth provider {kind}")))?;
|
||||
let client_id = provider.client_id()?;
|
||||
let json = credentials::get(credential_ref)?;
|
||||
let mut tokens: oauth::TokenSet =
|
||||
serde_json::from_str(&json).map_err(|e| SyncError::Credential(e.to_string()))?;
|
||||
if tokens.is_expired(now_unix()) {
|
||||
let refresh = tokens.refresh_token.clone().ok_or(SyncError::Auth)?;
|
||||
tokens = oauth::refresh_tokens(&provider, &client_id, &refresh).await?;
|
||||
let updated =
|
||||
serde_json::to_string(&tokens).map_err(|e| SyncError::Credential(e.to_string()))?;
|
||||
credentials::set(credential_ref, &updated)?;
|
||||
}
|
||||
Ok(tokens.access_token)
|
||||
}
|
||||
|
||||
/// Construct the concrete `SyncTarget` for a stored row by its `kind`. This is
|
||||
/// the one place that maps kind → provider impl (the pump/test dispatch here).
|
||||
#[cfg(feature = "sync")]
|
||||
pub struct BoxTarget;
|
||||
pub fn build_sync_target(
|
||||
row: &crate::storage::SyncTargetRow,
|
||||
) -> Result<Box<dyn SyncTarget>, SyncError> {
|
||||
match row.kind.as_str() {
|
||||
"webdav" => Ok(Box::new(WebDavTarget::from_row(row))),
|
||||
"onedrive" => Ok(Box::new(OneDriveTarget::from_row(row))),
|
||||
// Linking works (tokens stored); upload lands next, same trait.
|
||||
"dropbox" | "box" => Err(SyncError::Upload(format!(
|
||||
"{} upload is not implemented yet (OAuth linking works)",
|
||||
row.kind
|
||||
))),
|
||||
other => Err(SyncError::Upload(format!(
|
||||
"unknown sync target kind {other}"
|
||||
))),
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Secondary OAuth providers (Phase 9b) — same trait, separate impls ----
|
||||
|
||||
/// OneDrive via Microsoft Graph. The reference OAuth provider impl.
|
||||
#[cfg(feature = "sync")]
|
||||
pub struct OneDriveTarget {
|
||||
pub credential_ref: String,
|
||||
}
|
||||
|
||||
#[cfg(feature = "sync")]
|
||||
impl OneDriveTarget {
|
||||
pub fn from_row(row: &crate::storage::SyncTargetRow) -> Self {
|
||||
Self {
|
||||
credential_ref: row.credential_ref.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn bearer(&self) -> Result<String, SyncError> {
|
||||
resolve_access_token("onedrive", &self.credential_ref).await
|
||||
}
|
||||
|
||||
/// Graph path addressing: `…/root:/{path}{suffix}` (suffix `""` = the item,
|
||||
/// `:/content` = its bytes, `:/children` = its child collection).
|
||||
fn item_url(remote_path: &str, suffix: &str) -> String {
|
||||
let p = remote_path.trim_start_matches('/');
|
||||
format!("https://graph.microsoft.com/v1.0/me/drive/root:/{p}{suffix}")
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(feature = "sync")]
|
||||
#[async_trait]
|
||||
impl SyncTarget for OneDriveTarget {
|
||||
fn kind(&self) -> SyncKind {
|
||||
SyncKind::OneDrive
|
||||
}
|
||||
fn is_third_party(&self) -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
async fn test(&self) -> Result<(), SyncError> {
|
||||
let token = self.bearer().await?;
|
||||
let resp = http_client()?
|
||||
.get("https://graph.microsoft.com/v1.0/me/drive")
|
||||
.bearer_auth(token)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| SyncError::Unreachable(e.to_string()))?;
|
||||
match resp.status().as_u16() {
|
||||
code if (200..300).contains(&code) => Ok(()),
|
||||
401 | 403 => Err(SyncError::Auth),
|
||||
other => Err(SyncError::Unreachable(format!("graph {other}"))),
|
||||
}
|
||||
}
|
||||
|
||||
async fn ensure_dir(&self, remote_dir: &str) -> Result<(), SyncError> {
|
||||
let token = self.bearer().await?;
|
||||
let mut parent = String::new();
|
||||
for segment in remote_dir.split('/').filter(|s| !s.is_empty()) {
|
||||
let url = if parent.is_empty() {
|
||||
"https://graph.microsoft.com/v1.0/me/drive/root/children".to_string()
|
||||
} else {
|
||||
format!("https://graph.microsoft.com/v1.0/me/drive/root:/{parent}:/children")
|
||||
};
|
||||
let body = serde_json::json!({
|
||||
"name": segment,
|
||||
"folder": {},
|
||||
"@microsoft.graph.conflictBehavior": "replace",
|
||||
});
|
||||
let resp = http_client()?
|
||||
.post(&url)
|
||||
.bearer_auth(&token)
|
||||
.json(&body)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| SyncError::Upload(e.to_string()))?;
|
||||
match resp.status().as_u16() {
|
||||
200 | 201 | 409 => {} // created, or already exists
|
||||
401 | 403 => return Err(SyncError::Auth),
|
||||
other => return Err(SyncError::Upload(format!("mkdir {other}"))),
|
||||
}
|
||||
parent = if parent.is_empty() {
|
||||
segment.to_string()
|
||||
} else {
|
||||
format!("{parent}/{segment}")
|
||||
};
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn exists(&self, remote_path: &str, _sha256: &str) -> Result<bool, SyncError> {
|
||||
let token = self.bearer().await?;
|
||||
let resp = http_client()?
|
||||
.get(Self::item_url(remote_path, ""))
|
||||
.bearer_auth(token)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| SyncError::Unreachable(e.to_string()))?;
|
||||
Ok(resp.status().is_success())
|
||||
}
|
||||
|
||||
/// ponytail: simple content upload (Graph caps this at 250 MB). A resumable
|
||||
/// upload session for larger recordings is a follow-up.
|
||||
async fn put(
|
||||
&self,
|
||||
local: &Path,
|
||||
remote_path: &str,
|
||||
prog: ProgressSink,
|
||||
) -> Result<(), SyncError> {
|
||||
let token = self.bearer().await?;
|
||||
let bytes = tokio::fs::read(local)
|
||||
.await
|
||||
.map_err(|e| SyncError::Upload(e.to_string()))?;
|
||||
let total = bytes.len() as u64;
|
||||
let resp = http_client()?
|
||||
.put(Self::item_url(remote_path, ":/content"))
|
||||
.bearer_auth(token)
|
||||
.body(bytes)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| SyncError::Upload(e.to_string()))?;
|
||||
match resp.status().as_u16() {
|
||||
code if (200..300).contains(&code) => {
|
||||
let _ = prog.send((total, total));
|
||||
Ok(())
|
||||
}
|
||||
401 | 403 => Err(SyncError::Auth),
|
||||
other => Err(SyncError::Upload(format!("PUT {other}"))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user