///|
/// S3-compatible backend (AWS S3, Cloudflare R2, MinIO, Ceph).
///
/// The HTTP transport and the clock are injected rather than imported, so the
/// whole backend — including the status-code mapping that decides `Applied` /
/// `NotApplied` / `Unknown` — is testable without a network or a real bucket,
/// and so the same code can run on a native client or inside a Worker.

///|
/// Response of the injected transport: status, headers, body.
pub(all) struct HttpReply {
  status : Int
  headers : Map[String, String]
  body : Bytes
}

///|
pub struct S3Store {
  config : S3Config
  send : async (String, String, Map[String, String], Bytes) -> HttpReply raise @bit.GitError
  /// Returns the current time as `YYYYMMDDTHHMMSSZ`.
  now : () -> String
  /// Keys returned per listing page.
  page_size : Int
}

///|
pub fn S3Store::new(
  config : S3Config,
  send : async (String, String, Map[String, String], Bytes) -> HttpReply raise @bit.GitError,
  now : () -> String,
  page_size? : Int = 1000,
) -> S3Store {
  { config, send, now, page_size }
}

///|
/// Object path for a key, honouring path-style vs virtual-host addressing.
fn S3Store::object_path(self : S3Store, key : String) -> String {
  if self.config.path_style {
    "/\{self.config.bucket}/\{key}"
  } else {
    "/\{key}"
  }
}

///|
/// Bucket path, used by listings.
fn S3Store::bucket_path(self : S3Store) -> String {
  if self.config.path_style {
    "/\{self.config.bucket}"
  } else {
    "/"
  }
}

///|
/// Case-insensitive header lookup. Header casing varies by server, and R2 and
/// MinIO do not agree with AWS on it.
fn header_get(headers : Map[String, String], name : String) -> String? {
  let wanted = lowercase(name)
  for key, value in headers {
    if lowercase(key) == wanted {
      return Some(value)
    }
  }
  None
}

///|
/// S3 returns ETags wrapped in quotes; this API treats version tags as opaque
/// unquoted strings, so the quotes are stripped on the way in and restored on
/// the way out.
pub fn strip_etag_quotes(etag : String) -> String {
  let n = etag.length()
  if n >= 2 && etag[0] == '"' && etag[n - 1] == '"' {
    String::unsafe_substring(etag, start=1, end=n - 1)
  } else {
    etag
  }
}

///|
fn quote_etag(etag : String) -> String {
  if etag.length() >= 2 && etag[0] == '"' {
    etag
  } else {
    "\"\{etag}\""
  }
}

///|
/// Whether a status means the store never got a chance to decide.
///
/// A 5xx or a 429 leaves the write indeterminate: it may have been applied
/// before the failure. Reporting these as errors would tempt the caller into
/// a blind retry, which is how a WAL double-applies.
fn is_indeterminate(status : Int) -> Bool {
  status == 429 || status >= 500
}

///|
async fn S3Store::request(
  self : S3Store,
  verb : String,
  path : String,
  query : Array[(String, String)],
  headers : Map[String, String],
  body : Bytes,
) -> HttpReply raise @bit.GitError {
  let signed = sign_s3_request(
    self.config,
    verb,
    path,
    query,
    headers,
    body,
    (self.now)(),
  )
  (self.send)(signed.verb, signed.url, signed.headers, signed.body)
}

///|
fn reply_to_get(
  reply : HttpReply,
  key : String,
) -> GetOutcome raise @bit.GitError {
  if reply.status == 304 {
    return GetOutcome::NotModified
  }
  if reply.status == 404 {
    return GetOutcome::Missing
  }
  if reply.status == 200 || reply.status == 206 {
    let etag = match header_get(reply.headers, "etag") {
      Some(value) => strip_etag_quotes(value)
      None => ""
    }
    return GetOutcome::Found(reply.body, etag)
  }
  raise @bit.GitError::IoError("GET \{key} failed: HTTP \{reply.status}")
}

///|
pub impl ObjectStore for S3Store with fn get(self, key) {
  let reply = self.request(
    "GET",
    self.object_path(key),
    [],
    Map([]),
    Bytes::new(0),
  )
  reply_to_get(reply, key)
}

///|
pub impl ObjectStore for S3Store with fn get_range(self, key, range) {
  let headers : Map[String, String] = {
    "Range": "bytes=\{range.start}-\{range.end}",
  }
  let reply = self.request(
    "GET",
    self.object_path(key),
    [],
    headers,
    Bytes::new(0),
  )
  reply_to_get(reply, key)
}

///|
pub impl ObjectStore for S3Store with fn get_if_none_match(self, key, etag) {
  let headers : Map[String, String] = { "If-None-Match": quote_etag(etag) }
  let reply = self.request(
    "GET",
    self.object_path(key),
    [],
    headers,
    Bytes::new(0),
  )
  reply_to_get(reply, key)
}

///|
pub impl ObjectStore for S3Store with fn put(self, key, body, condition) {
  let headers : Map[String, String] = Map([])
  match condition {
    Unconditional => ()
    IfNotExists => headers["If-None-Match"] = "*"
    IfMatch(etag) => headers["If-Match"] = quote_etag(etag)
  }
  let reply = self.request("PUT", self.object_path(key), [], headers, body) catch {
    // A transport failure after the request left the client is exactly the
    // indeterminate case: the object may be there.
    _ => return PutOutcome::Unknown
  }
  if reply.status == 200 || reply.status == 201 {
    match header_get(reply.headers, "etag") {
      Some(value) => PutOutcome::Applied(strip_etag_quotes(value))
      // Without an ETag we cannot name the version we just wrote, and
      // guessing one would poison the next compare-and-swap. Make the caller
      // re-read instead.
      None => PutOutcome::Unknown
    }
  } else if reply.status == 412 || reply.status == 409 {
    // 412: the precondition did not hold.
    // 409: a concurrent conditional write won. Both mean "re-read and retry".
    PutOutcome::NotApplied
  } else if is_indeterminate(reply.status) {
    PutOutcome::Unknown
  } else if reply.status == 400 && condition is IfMatch(_) {
    // Some S3-compatible stores answer 400 rather than 501 for an
    // unsupported If-Match. Surface it as a hard error: silently treating it
    // as NotApplied would loop forever.
    raise @bit.GitError::IoError(
      "PUT \{key}: store rejected the conditional write (HTTP 400); " +
      "this backend may not support compare-and-swap",
    )
  } else {
    raise @bit.GitError::IoError("PUT \{key} failed: HTTP \{reply.status}")
  }
}

///|
pub impl ObjectStore for S3Store with fn delete(self, key) {
  let reply = self.request(
    "DELETE",
    self.object_path(key),
    [],
    Map([]),
    Bytes::new(0),
  )
  // 404 is success: the key is gone, which is what was asked for.
  if reply.status == 200 ||
    reply.status == 204 ||
    reply.status == 202 ||
    reply.status == 404 {
    return
  }
  raise @bit.GitError::IoError("DELETE \{key} failed: HTTP \{reply.status}")
}

///|
pub impl ObjectStore for S3Store with fn list(self, prefix, after) {
  let query : Array[(String, String)] = [
    ("list-type", "2"),
    ("max-keys", self.page_size.to_string()),
    ("prefix", prefix),
  ]
  if after != "" {
    query.push(("start-after", after))
  }
  let reply = self.request(
    "GET",
    self.bucket_path(),
    query,
    Map([]),
    Bytes::new(0),
  )
  if reply.status != 200 {
    raise @bit.GitError::IoError("LIST \{prefix} failed: HTTP \{reply.status}")
  }
  parse_list_objects_v2(@utf8.decode_lossy(reply.body[:]))
}

///|
pub impl ObjectStore for S3Store with fn supports_cas(_self) {
  true
}