194c44c
use std::collections::HashMap;
use std::pin::Pin;
use std::task::Poll;
use export::storage::AppStorage;
use models::git::git_lfs;
use models::repo::repos;
use sea_orm::sea_query::Expr;
use sea_orm::{ColumnTrait, EntityTrait, ExprTrait, QueryFilter, Set};
use sha2::{Digest, Sha256};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::lfs::types::{
LfsAction, LfsBatchObject, LfsBatchRequest, LfsBatchResponse, LfsError, LfsOperation,
LfsVerifyRequest,
};
use crate::lfs::validate::{canonical_oid, validate_size};
use crate::service::GitService;
use crate::share::errors::GitError;
pub const UPLOAD_TTL_SECS: u32 = 3600;
pub const DOWNLOAD_TTL_SECS: u32 = 3600;
pub const MAX_BATCH_OBJECTS: usize = 200;
pub async fn batch(
svc: &GitService,
space_name: String,
repo_name: String,
req: LfsBatchRequest,
) -> Result<LfsBatchResponse, GitError> {
if req.objects.len() > MAX_BATCH_OBJECTS {
return Err(GitError::InvalidArgument(format!(
"batch request exceeds {MAX_BATCH_OBJECTS} objects"
)));
}
let ctx = svc.context(space_name.clone(), repo_name.clone()).await?;
if !ctx.model.enable_lfs {
return Err(GitError::NotFound);
}
let storage: &AppStorage = &svc.state.storage;
let cap = svc.state.config.git.max_blob_size;
let mut out: Vec<LfsBatchObject> = Vec::with_capacity(req.objects.len());
for obj in req.objects {
let oid = match canonical_oid(&obj.oid) {
Ok(oid) => oid,
Err(e) => {
out.push(object_error(obj.oid, obj.size, e));
continue;
}
};
if let Err(e) = validate_size(obj.size, cap) {
out.push(object_error(oid, obj.size, e));
continue;
}
let present = storage
.client()
.exists(&storage_key(&ctx.model.id, &oid))
.await
.map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;
let actions = build_actions(
svc,
&ctx.model,
&space_name,
&repo_name,
&oid,
req.operation,
present,
)?;
let mut resp = LfsBatchObject {
oid,
size: obj.size,
actions,
error: None,
};
if req.operation == LfsOperation::Download && !present {
resp.error = Some(LfsError {
code: 404,
message: "object not found".to_string(),
});
}
out.push(resp);
}
Ok(LfsBatchResponse {
transfer: "basic".to_string(),
hash_algo: "sha256".to_string(),
objects: out,
})
}
fn build_actions(
svc: &GitService,
repo: &repos::Model,
space_name: &str,
repo_name: &str,
oid: &str,
operation: LfsOperation,
present: bool,
) -> Result<HashMap<String, LfsAction>, GitError> {
let storage: &AppStorage = &svc.state.storage;
let mut actions: HashMap<String, LfsAction> = HashMap::new();
match (operation, present) {
(LfsOperation::Download, true) | (LfsOperation::Verify, true) => {
actions.insert(
"download".to_string(),
LfsAction {
href: storage
.client()
.presign_get(&storage_key(&repo.id, oid), DOWNLOAD_TTL_SECS)
.map_err(|e| GitError::Internal(format!("presign_get: {e}")))?,
header: None,
expires_at: Some(now_plus_secs(DOWNLOAD_TTL_SECS as i64)),
},
);
}
(LfsOperation::Upload, _) => {
let href = storage
.client()
.presign_put(&storage_key(&repo.id, oid), UPLOAD_TTL_SECS)
.map_err(|e| GitError::Internal(format!("presign_put: {e}")))?;
actions.insert(
"upload".to_string(),
LfsAction {
href: href.clone(),
header: None,
expires_at: Some(now_plus_secs(UPLOAD_TTL_SECS as i64)),
},
);
if !present {
actions.insert(
"verify".to_string(),
LfsAction {
href: verify_href(svc, space_name, repo_name),
header: None,
expires_at: Some(now_plus_secs(UPLOAD_TTL_SECS as i64)),
},
);
}
}
_ => {}
}
Ok(actions)
}
/// Stream an object upload: the reader is consumed chunk-by-chunk into
/// storage while its bytes are hashed with SHA-256 on the fly. The hash is
/// compared against the canonical oid only after the upload finishes; a
/// mismatch (or declared-size mismatch) deletes the stored object again.
/// Uploads are idempotent (already-present objects succeed without a second
/// write). `repos.lfs_size` is incremented atomically.
#[allow(clippy::too_many_arguments)]
pub async fn upload_stream<R>(
svc: &GitService,
space_name: String,
repo_name: String,
caller: i64,
oid: String,
size: i64,
content_type: Option<String>,
reader: R,
) -> Result<bool, GitError>
where
R: tokio::io::AsyncRead + Unpin + Send,
{
let _ = caller;
let ctx = svc.context(space_name, repo_name).await?;
if !ctx.model.enable_lfs {
return Err(GitError::NotFound);
}
let oid = canonical_oid(&oid)?;
validate_size(size, svc.state.config.git.max_blob_size)?;
let storage: &AppStorage = &svc.state.storage;
let key = storage_key(&ctx.model.id, &oid);
let existed = storage
.client()
.exists(&key)
.await
.map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;
if !existed {
let mut reader = HashingReader::new(tokio::io::AsyncReadExt::take(reader, size as u64 + 1));
let sent = storage
.client()
.put_stream(
&key,
&mut reader,
content_type
.as_deref()
.unwrap_or("application/octet-stream"),
)
.await
.map_err(|e| GitError::Internal(format!("lfs put_stream: {e}")))?;
let (hex, _count) = reader.finish();
if sent as i64 != size {
let _ = storage.client().delete(&key).await;
return Err(GitError::InvalidArgument(format!(
"size mismatch: declared {size}, received {sent}"
)));
}
let actual = format!("sha256:{hex}");
if actual != oid {
let _ = storage.client().delete(&key).await;
return Err(GitError::InvalidArgument(format!(
"oid mismatch: expected {oid}, got {actual}"
)));
}
}
match git_lfs::Entity::insert(git_lfs::ActiveModel {
id: Set(Uuid::now_v7()),
repo_id: Set(ctx.model.id),
oid: Set(oid),
size: Set(size),
content_type: Set(content_type),
created_at: Set(OffsetDateTime::now_utc()),
})
.exec(svc.db_write())
.await
{
Ok(_) => {}
Err(db) => match GitError::from(db) {
GitError::Conflict(_) => {}
other => return Err(other),
},
}
if !existed {
let update = repos::Entity::update_many()
.col_expr(
repos::Column::LfsSize,
Expr::col(repos::Column::LfsSize).add(size),
)
.col_expr(
repos::Column::UpdatedAt,
Expr::val(OffsetDateTime::now_utc()),
)
.filter(repos::Column::Id.eq(ctx.model.id));
let res = if let Some(cap) = ctx.model.storage_limit_size {
let limit = cap.saturating_sub(size);
update
.filter(Expr::col(repos::Column::LfsSize).lte(limit))
.exec(svc.db_write())
.await?
} else {
update.exec(svc.db_write()).await?
};
if res.rows_affected == 0 {
let _ = storage.client().delete(&key).await;
return Err(GitError::InvalidArgument(
"lfs storage limit exceeded".to_string(),
));
}
}
Ok(true)
}
/// In-memory convenience wrapper over [`upload_stream`] for small bodies.
#[allow(clippy::too_many_arguments)]
pub async fn upload(
svc: &GitService,
space_name: String,
repo_name: String,
caller: i64,
oid: String,
size: i64,
content_type: Option<String>,
body: Vec<u8>,
) -> Result<bool, GitError> {
upload_stream(
svc,
space_name,
repo_name,
caller,
oid,
size,
content_type,
std::io::Cursor::new(body),
)
.await
}
/// `POST objects/verify` — confirm a stored object matches the recorded
/// `(oid, size)`; fails with `NotFound` when the object or its DB row is
/// missing.
pub async fn verify(
svc: &GitService,
space_name: String,
repo_name: String,
req: LfsVerifyRequest,
) -> Result<(), GitError> {
let oid = canonical_oid(&req.oid)?;
let ctx = svc.context(space_name, repo_name).await?;
let row = git_lfs::Entity::find()
.filter(git_lfs::Column::RepoId.eq(ctx.model.id))
.filter(git_lfs::Column::Oid.eq(&oid))
.one(svc.db_read())
.await?
.ok_or(GitError::NotFound)?;
if row.size != req.size {
return Err(GitError::InvalidArgument("size mismatch".to_string()));
}
let storage: &AppStorage = &svc.state.storage;
let present = storage
.client()
.exists(&storage_key(&ctx.model.id, &oid))
.await
.map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;
if !present {
return Err(GitError::NotFound);
}
Ok(())
}
/// Stream an object into `writer` (bounded memory, no whole-object buffering).
/// Returns the number of bytes written.
pub async fn download_to<W>(
svc: &GitService,
space_name: String,
repo_name: String,
oid: String,
writer: &mut W,
) -> Result<u64, GitError>
where
W: tokio::io::AsyncWrite + Send + Unpin,
{
let oid = canonical_oid(&oid)?;
let ctx = svc.context(space_name, repo_name).await?;
let storage: &AppStorage = &svc.state.storage;
storage
.client()
.get_to_writer(&storage_key(&ctx.model.id, &oid), writer)
.await
.map_err(|e| GitError::Internal(format!("lfs get: {e}")))
}
/// Whole-object convenience wrapper over [`download_to`] (use only for small
/// objects; prefer the streaming variant in hot paths).
pub async fn download(
svc: &GitService,
space_name: String,
repo_name: String,
oid: String,
) -> Result<Vec<u8>, GitError> {
let mut buf = Vec::new();
download_to(svc, space_name, repo_name, oid, &mut buf).await?;
Ok(buf)
}
/// Consistency repair + unreferenced-object sweep for one repo:
///
/// 1. drop `git_lfs` rows whose storage object has vanished;
/// 2. delete storage keys under `lfs/{repo}/` that have no DB row.
///
/// Returns the number of removed entries.
pub async fn cleanup_unreferenced(
svc: &GitService,
space_name: String,
repo_name: String,
) -> Result<i64, GitError> {
let ctx = svc.context(space_name, repo_name).await?;
let repo_id = ctx.model.id;
let storage: &AppStorage = &svc.state.storage;
let mut removed: i64 = 0;
let db = svc.db_write();
let rows = git_lfs::Entity::find()
.filter(git_lfs::Column::RepoId.eq(repo_id))
.all(db)
.await?;
for row in rows {
let key = storage_key(&repo_id, &row.oid);
let exists = storage
.client()
.exists(&key)
.await
.map_err(|e| GitError::Internal(format!("storage exists: {e}")))?;
if !exists {
git_lfs::Entity::delete_by_id(row.id).exec(db).await?;
removed += 1;
}
}
let prefix = format!("lfs/{}/", repo_id.simple());
let live: std::collections::HashSet<String> = git_lfs::Entity::find()
.filter(git_lfs::Column::RepoId.eq(repo_id))
.all(svc.db_read())
.await?
.into_iter()
.map(|r| r.oid)
.collect();
let mut token: Option<String> = None;
loop {
let page = storage
.client()
.list_page(&prefix, token.as_deref(), Some(200))
.await
.map_err(|e| GitError::Internal(format!("storage list: {e}")))?;
for obj in page.objects {
let Some(oid) = obj.key.strip_prefix(&prefix) else {
continue;
};
if !live.contains(oid) {
storage
.client()
.delete(&obj.key)
.await
.map_err(|e| GitError::Internal(format!("storage delete: {e}")))?;
removed += 1;
}
}
match page.next_continuation_token {
Some(next) if !next.is_empty() => token = Some(next),
_ => break,
}
}
Ok(removed)
}
fn object_error(oid: String, size: i64, err: GitError) -> LfsBatchObject {
LfsBatchObject {
oid,
size,
actions: HashMap::new(),
error: Some(LfsError {
code: crate::lfs::error::http_status(&err) as i32,
message: err.to_string(),
}),
}
}
fn storage_key(repo_id: &Uuid, oid: &str) -> String {
format!("lfs/{}/{}", repo_id.simple(), oid)
}
/// Tee between the upload reader and storage: every byte read is fed into a
/// running SHA-256 and counted, so the upload can be verified without
/// buffering the whole body.
struct HashingReader<R> {
inner: R,
hasher: Sha256,
count: u64,
}
impl<R> HashingReader<R> {
fn new(inner: R) -> Self {
Self {
inner,
hasher: Sha256::new(),
count: 0,
}
}
fn finish(self) -> (String, u64) {
let hex = self
.hasher
.finalize()
.iter()
.map(|b| format!("{b:02x}"))
.collect();
(hex, self.count)
}
}
impl<R: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for HashingReader<R> {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
let before = buf.filled().len();
let poll = Pin::new(&mut self.inner).poll_read(cx, buf);
if poll.is_ready() {
let written = &buf.filled()[before..];
if !(written.is_empty() && buf.remaining() > 0) {
self.hasher.update(written);
self.count += written.len() as u64;
}
}
poll
}
}
fn now_plus_secs(secs: i64) -> String {
OffsetDateTime::now_utc()
.saturating_add(time::Duration::seconds(secs))
.format(&time::format_description::well_known::Rfc3339)
.unwrap_or_default()
}
/// The batch `verify` action always points back at this server (the initial
/// upload itself goes directly to the presigned S3 URL). The href follows the
/// Gitea-style `/{space}/{repo}.git/info/lfs` layout, prefixed with
/// `web.web_base_url` when configured.
fn verify_href(svc: &GitService, space_name: &str, repo_name: &str) -> String {
let base = svc.state.config.web.web_base_url.trim_end_matches('/');
if base.is_empty() {
format!("/{}/{}.git/info/lfs/objects/verify", space_name, repo_name)
} else {
format!(
"{base}/{}/{}.git/info/lfs/objects/verify",
space_name, repo_name
)
}
}