994dbc3690
Chips: Neutral / Sarcastic / Funny / Dry wit. The chosen voice rides
the reprocess call and the button echoes it ('Re-summarize · Sarcastic');
an existing summary shows '· in a sarcastic voice' next to its heading.
compileKotlin + 39 unit tests green.
753 lines
33 KiB
Kotlin
753 lines
33 KiB
Kotlin
package com.shonar.provider
|
|
|
|
import java.io.File
|
|
import java.security.MessageDigest
|
|
import java.time.Instant
|
|
import java.util.concurrent.TimeUnit
|
|
import kotlinx.coroutines.Dispatchers
|
|
import kotlinx.coroutines.ensureActive
|
|
import kotlinx.coroutines.flow.MutableStateFlow
|
|
import kotlinx.coroutines.flow.StateFlow
|
|
import kotlinx.coroutines.sync.Mutex
|
|
import kotlinx.coroutines.sync.withLock
|
|
import kotlinx.coroutines.withContext
|
|
import okhttp3.MediaType.Companion.toMediaType
|
|
import okhttp3.OkHttpClient
|
|
import okhttp3.Request
|
|
import okhttp3.RequestBody.Companion.toRequestBody
|
|
import okhttp3.Response
|
|
|
|
/**
|
|
* P3: SHONAR-backend provider (this repo's FastAPI server, M1 auth + M2
|
|
* uploads). Speaks only documented endpoints:
|
|
*
|
|
* - GET /api/v1/provider-info (probe; no credentials)
|
|
* - POST /api/v1/auth/login {email, password, device_name, platform}
|
|
* - POST /api/v1/auth/refresh {refresh_token} (rotating; reuse-detected)
|
|
* - POST /api/v1/auth/logout {refresh_token}
|
|
* - GET /api/v1/auth/me (token validation)
|
|
* - POST /api/v1/auth/delete-account {password} (via [deleteAccount])
|
|
* - POST /api/v1/uploads (create session)
|
|
* - GET /api/v1/uploads/{id} (resume: received indexes)
|
|
* - PUT /api/v1/uploads/{id}/chunks/{n} (+ X-Chunk-Sha256)
|
|
* - POST /api/v1/uploads/{id}/finalize
|
|
* - GET /api/v1/recordings?limit&offset&sort&order
|
|
* - GET /api/v1/recordings/{id}/audio
|
|
* - DELETE /api/v1/recordings/{id}?purge=true
|
|
*
|
|
* Token discipline: the backend rotates refresh tokens with reuse
|
|
* detection, so the stored pair is overwritten on every login AND every
|
|
* refresh, and concurrent 401s serialize on [refreshMutex] — two parallel
|
|
* refreshes would look like token reuse and burn the whole family.
|
|
*
|
|
* Sidecars: file cache under [sidecarRoot] keyed by remote recording id.
|
|
* AI content (transcript/summary/jobs) goes through the real M7
|
|
* endpoints instead — see [fetchTranscript] and friends, consumed by
|
|
* the details screen.
|
|
*
|
|
* Cancellation safety: a cancelled upload leaves an open server session
|
|
* with some chunks stored — invisible until finalize, resumable via the
|
|
* status endpoint, and idempotent per draft id through
|
|
* `client_recording_id`. Nothing half-visible ever appears in listings.
|
|
*/
|
|
class CustomShonarProvider(
|
|
private val auth: ShonarAuthStore,
|
|
private val sidecarRoot: File,
|
|
private val client: OkHttpClient = defaultClient(),
|
|
private val handshake: ShonarHandshake = ShonarHandshake(),
|
|
/** Identifies this client to the server (login device_name/platform). */
|
|
private val deviceName: String = "SHONAR Desktop",
|
|
private val platform: String = "desktop",
|
|
) : ShonarProvider {
|
|
|
|
override val descriptor = ProviderDescriptor(
|
|
id = ProviderRegistry.CUSTOM_SHONAR_ID,
|
|
displayName = "Custom SHONAR server",
|
|
capabilities = setOf(
|
|
ProviderDescriptor.Capability.CHUNKED_UPLOAD,
|
|
ProviderDescriptor.Capability.ACCOUNT_DELETION,
|
|
),
|
|
)
|
|
|
|
private val _authState = MutableStateFlow(AuthState.DISCONNECTED)
|
|
override val authState: StateFlow<AuthState> = _authState
|
|
|
|
/** Base origin, e.g. https://shonar.example.com. Set by connect/reconnect. */
|
|
private var origin: String? = null
|
|
|
|
private val refreshMutex = Mutex()
|
|
|
|
// ---- lifecycle ---------------------------------------------------------
|
|
|
|
override suspend fun probe(baseUrl: ServerUrl): ProbeResult = handshake.probe(baseUrl)
|
|
|
|
override suspend fun connect(credential: ProviderCredential) = withContext(Dispatchers.IO) {
|
|
when (credential) {
|
|
is ProviderCredential.ShonarLogin -> login(
|
|
credential.serverUrl.origin, credential.email, credential.password
|
|
)
|
|
is ProviderCredential.OAuthTokens -> resumeWithTokens(credential)
|
|
else -> throw ProviderError.InvalidUrl(
|
|
"Custom SHONAR server needs an email + password login"
|
|
)
|
|
}
|
|
}
|
|
|
|
override suspend fun reconnect(): AuthState = withContext(Dispatchers.IO) {
|
|
val saved = auth.load()
|
|
if (saved == null || saved.refreshToken.isBlank()) {
|
|
_authState.value = AuthState.DISCONNECTED
|
|
return@withContext _authState.value
|
|
}
|
|
origin = saved.baseUrl
|
|
val base = saved.baseUrl
|
|
try {
|
|
executeAuthed(base) { token -> get(base, "/api/v1/auth/me", token) }.use { resp ->
|
|
if (resp.code != 200) throw ProviderError.AuthExpired()
|
|
}
|
|
_authState.value = AuthState.CONNECTED
|
|
} catch (e: ProviderError.AuthExpired) {
|
|
// Refresh already failed inside executeAuthed: tokens are dead.
|
|
auth.clearTokens()
|
|
_authState.value = AuthState.EXPIRED
|
|
} catch (e: ProviderError) {
|
|
_authState.value = AuthState.OFFLINE
|
|
}
|
|
_authState.value
|
|
}
|
|
|
|
override suspend fun disconnect(revokeOnServer: Boolean) = withContext(Dispatchers.IO) {
|
|
if (revokeOnServer) {
|
|
// Best effort: local state is cleared even if revoke fails.
|
|
runCatching {
|
|
val saved = auth.load()
|
|
if (saved != null && saved.refreshToken.isNotBlank()) {
|
|
postUnauthed(
|
|
saved.baseUrl, "/api/v1/auth/logout",
|
|
"""{"refresh_token":${jsonStr(saved.refreshToken)}}""",
|
|
).close()
|
|
}
|
|
}
|
|
}
|
|
auth.clearTokens()
|
|
_authState.value = AuthState.DISCONNECTED
|
|
}
|
|
|
|
override suspend fun deleteAccountAndData() {
|
|
// No password is available here (never stored), so server-side
|
|
// account purge needs the explicit [deleteAccount] call below.
|
|
// This path revokes the session and wipes everything local.
|
|
withContext(Dispatchers.IO) {
|
|
disconnect(revokeOnServer = true)
|
|
sidecarRoot.deleteRecursively()
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Full server-side account purge (backend: 30-day grace, then hard
|
|
* delete). Needs the password because it is never stored — call this
|
|
* from a confirmation screen that asks for it once.
|
|
*/
|
|
suspend fun deleteAccount(password: String) = withContext(Dispatchers.IO) {
|
|
val base = origin ?: auth.load()?.baseUrl ?: throw ProviderError.NotConnected()
|
|
executeAuthed(base) { token ->
|
|
post(base, "/api/v1/auth/delete-account", token,
|
|
"""{"password":${jsonStr(password)}}""")
|
|
}.use { resp ->
|
|
if (resp.code != 202 && resp.code != 204) {
|
|
throw ProviderError.Transient("Account deletion refused (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
auth.clearAll()
|
|
origin = null
|
|
sidecarRoot.deleteRecursively()
|
|
_authState.value = AuthState.DISCONNECTED
|
|
}
|
|
|
|
// ---- storage -----------------------------------------------------------
|
|
|
|
override suspend fun upload(draft: RecordingDraft, onProgress: (Float) -> Unit): RemoteRef =
|
|
withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
val total = draft.sizeBytes.coerceAtLeast(1)
|
|
|
|
val (sessionId, chunkSize) = createSession(base, draft)
|
|
val received = uploadStatus(base, sessionId).toMutableSet()
|
|
|
|
var sent = 0L
|
|
// Account progress for already-present chunks so resume continues
|
|
// the bar instead of restarting it.
|
|
var idx = 0
|
|
val chunkCount = ((draft.sizeBytes + chunkSize - 1) / chunkSize).toInt().coerceAtLeast(1)
|
|
var resumedBytes = 0L
|
|
// Estimate resumed bytes from the received set (last chunk may be short).
|
|
for (i in received) {
|
|
resumedBytes += if (i < chunkCount - 1) chunkSize.toLong()
|
|
else (draft.sizeBytes - chunkSize * (chunkCount - 1)).coerceAtLeast(0)
|
|
}
|
|
sent = resumedBytes
|
|
if (sent > 0) onProgress((sent.toFloat() / total).coerceIn(0f, 1f))
|
|
|
|
while (idx < chunkCount) {
|
|
// Cooperative cancellation between chunks; a cancelled
|
|
// upload leaves a resumable server session, never a
|
|
// half-visible object.
|
|
ensureActive()
|
|
if (idx !in received) {
|
|
val slice = readSlice(draft.sourceFile, idx.toLong() * chunkSize, chunkSize)
|
|
putChunk(base, sessionId, idx, slice)
|
|
sent += slice.size
|
|
onProgress((sent.toFloat() / total).coerceIn(0f, 1f))
|
|
}
|
|
idx++
|
|
}
|
|
|
|
val recordingId = finalize(base, sessionId, draft)
|
|
onProgress(1f)
|
|
RemoteRef(ProviderRegistry.CUSTOM_SHONAR_ID, recordingId, etag = null, sizeBytes = draft.sizeBytes)
|
|
}
|
|
|
|
override suspend fun download(ref: RemoteRef, dest: File, onProgress: (Float) -> Unit) =
|
|
withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
executeAuthed(base) { token -> get(base, "/api/v1/recordings/${ref.key}/audio", token) }.use { resp ->
|
|
when (resp.code) {
|
|
200 -> {
|
|
val body = resp.body ?: throw ProviderError.Transient("Empty download response")
|
|
val total = body.contentLength().takeIf { it > 0 } ?: -1
|
|
dest.parentFile?.mkdirs()
|
|
body.byteStream().use { input ->
|
|
dest.outputStream().use { output ->
|
|
val buf = ByteArray(64 * 1024)
|
|
var written = 0L
|
|
var lastReported = -1f
|
|
while (true) {
|
|
val n = input.read(buf)
|
|
if (n < 0) break
|
|
output.write(buf, 0, n)
|
|
written += n
|
|
if (total > 0) {
|
|
val p = (written.toFloat() / total).coerceIn(0f, 1f)
|
|
if (p > lastReported) {
|
|
onProgress(p)
|
|
lastReported = p
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
onProgress(1f)
|
|
}
|
|
404 -> throw ProviderError.NotFound(ref.key)
|
|
else -> throw ProviderError.Transient("Download failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
override suspend fun delete(ref: RemoteRef) {
|
|
withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
executeAuthed(base) { token ->
|
|
Request.Builder().url("$base/api/v1/recordings/${ref.key}?purge=true")
|
|
.delete().header("Authorization", "Bearer $token").build()
|
|
}.use { resp ->
|
|
when (resp.code) {
|
|
204, 200 -> {
|
|
sidecarDir(ref).deleteRecursively()
|
|
}
|
|
404 -> throw ProviderError.NotFound(ref.key)
|
|
else -> throw ProviderError.Transient("Delete failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
override suspend fun list(cursor: String?): Page<RemoteRecording> = withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
val offset = cursor?.toIntOrNull()?.coerceAtLeast(0) ?: 0
|
|
val limit = 200
|
|
executeAuthed(base) { token ->
|
|
get(base, "/api/v1/recordings?limit=$limit&offset=$offset&sort=recorded_at&order=desc", token)
|
|
}.use { resp ->
|
|
if (resp.code != 200) throw ProviderError.Transient("Listing failed (HTTP ${resp.code})")
|
|
val root = org.json.JSONObject(resp.body?.string().orEmpty())
|
|
val total = root.optInt("total", 0)
|
|
val items = root.optJSONArray("items") ?: org.json.JSONArray()
|
|
val out = mutableListOf<RemoteRecording>()
|
|
for (i in 0 until items.length()) {
|
|
parseRecording(items.getJSONObject(i))?.let { out += it }
|
|
}
|
|
val next = if (offset + limit < total) (offset + limit).toString() else null
|
|
Page(out, next)
|
|
}
|
|
}
|
|
|
|
// ---- AI content (M7 endpoints, consumed by the M8 details screen) ----
|
|
//
|
|
// The recording id here is the server-side UUID — the local row's
|
|
// [com.shonar.recording.RecordingEntity.remoteKey] after upload.
|
|
// 404 surfaces as [ProviderError.NotFound] ("no transcript yet" is a
|
|
// normal state, not a failure). Raw JSON comes back so parsing stays in
|
|
// the testable `recording.AiContent` helpers, not in this IO class.
|
|
|
|
suspend fun fetchTranscript(recordingId: String): String = getAiJson(
|
|
"/api/v1/recordings/$recordingId/transcript", recordingId
|
|
)
|
|
|
|
suspend fun fetchSummary(recordingId: String): String = getAiJson(
|
|
"/api/v1/recordings/$recordingId/summary", recordingId
|
|
)
|
|
|
|
suspend fun fetchJobs(recordingId: String): String = getAiJson(
|
|
"/api/v1/recordings/$recordingId/jobs", recordingId
|
|
)
|
|
|
|
/** Force one pipeline stage to re-run on the server (no re-upload).
|
|
* [model] only applies to job="transcribe" (switches the saved
|
|
* per-recording override). 409 = stage already running or nothing to
|
|
* summarize; surfaced as ProviderError.Transient with the server's
|
|
* detail message. */
|
|
suspend fun reprocess(recordingId: String, job: String, model: String? = null,
|
|
tone: String? = null): String =
|
|
withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
val q = buildString {
|
|
if (model != null) append("&model=$model")
|
|
if (tone != null)
|
|
append("&tone=").append(java.net.URLEncoder.encode(tone, "UTF-8"))
|
|
}
|
|
executeAuthed(base) { token ->
|
|
post(base, "/api/v1/recordings/$recordingId/reprocess?job=$job$q", token, "{}")
|
|
}.use { resp ->
|
|
when (resp.code) {
|
|
200, 201 -> resp.body?.string().orEmpty()
|
|
404 -> throw ProviderError.NotFound(recordingId)
|
|
else -> {
|
|
val body = runCatching {
|
|
org.json.JSONObject(resp.body?.string().orEmpty())
|
|
.optString("detail")
|
|
}.getOrNull()
|
|
throw ProviderError.Transient(
|
|
body?.takeIf { it.isNotBlank() }
|
|
?: "Reprocess failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/** User edit → new server version with edited_by_user=true (pipeline won't clobber). */
|
|
suspend fun updateTranscript(
|
|
recordingId: String,
|
|
payloadJson: String,
|
|
): String = withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
executeAuthed(base) { token -> put(base, "/api/v1/recordings/$recordingId/transcript", token, payloadJson) }
|
|
.use { resp ->
|
|
when (resp.code) {
|
|
200 -> resp.body?.string().orEmpty()
|
|
404 -> throw ProviderError.NotFound(recordingId)
|
|
else -> throw ProviderError.Transient("Transcript update failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
suspend fun updateSummary(recordingId: String, payloadJson: String): String =
|
|
withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
executeAuthed(base) { token -> put(base, "/api/v1/recordings/$recordingId/summary", token, payloadJson) }
|
|
.use { resp ->
|
|
when (resp.code) {
|
|
200 -> resp.body?.string().orEmpty()
|
|
404 -> throw ProviderError.NotFound(recordingId)
|
|
else -> throw ProviderError.Transient("Summary update failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Title/notes rename from the details screen (server + Room updated by caller). */
|
|
suspend fun patchRecording(recordingId: String, payloadJson: String): String =
|
|
withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
executeAuthed(base) { token -> patch(base, "/api/v1/recordings/$recordingId", token, payloadJson) }
|
|
.use { resp ->
|
|
when (resp.code) {
|
|
200 -> resp.body?.string().orEmpty()
|
|
404 -> throw ProviderError.NotFound(recordingId)
|
|
else -> throw ProviderError.Transient("Recording update failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
private suspend fun getAiJson(path: String, recordingId: String): String =
|
|
withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
executeAuthed(base) { token -> get(base, path, token) }.use { resp ->
|
|
when (resp.code) {
|
|
200 -> resp.body?.string().orEmpty()
|
|
404 -> throw ProviderError.NotFound(recordingId)
|
|
else -> throw ProviderError.Transient("Request failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---- transcription models (Stage 1 endpoints, desktop Settings) --------
|
|
// Raw JSON throughout; parsing lives in the testable AiContent helpers.
|
|
|
|
suspend fun fetchModels(): String = withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
executeAuthed(base) { token -> get(base, "/api/v1/models", token) }.use { resp ->
|
|
if (resp.code != 200) throw ProviderError.Transient("Models request failed (HTTP ${resp.code})")
|
|
resp.body?.string().orEmpty()
|
|
}
|
|
}
|
|
|
|
suspend fun setDefaultModel(model: String): String = withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
val payload = """{"model":${jsonStr(model)}}"""
|
|
executeAuthed(base) { token -> put(base, "/api/v1/models/default", token, payload) }
|
|
.use { resp ->
|
|
when (resp.code) {
|
|
200 -> resp.body?.string().orEmpty()
|
|
422 -> throw ProviderError.InvalidUrl("Unsupported model: $model")
|
|
else -> throw ProviderError.Transient("Default model update failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
/** Fetch a model into the server cache (needs internet once; may take minutes). */
|
|
suspend fun downloadModel(model: String): String = withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
executeAuthed(base) { token ->
|
|
post(base, "/api/v1/models/$model/download", token, "{}")
|
|
}.use { resp ->
|
|
when (resp.code) {
|
|
200 -> resp.body?.string().orEmpty()
|
|
422 -> throw ProviderError.InvalidUrl("Unsupported model: $model")
|
|
501 -> throw ProviderError.Transient("faster-whisper is not installed on the server")
|
|
else -> throw ProviderError.Transient("Model download failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---- sidecars (local cache until the backend gains endpoints in M7) ----
|
|
|
|
private fun sidecarDir(ref: RemoteRef) = File(sidecarRoot, ref.key)
|
|
|
|
override suspend fun putSidecar(ref: RemoteRef, kind: SidecarKind, bytes: ByteArray) =
|
|
withContext(Dispatchers.IO) {
|
|
val f = File(sidecarDir(ref), kind.fileName)
|
|
f.parentFile?.mkdirs()
|
|
f.writeBytes(bytes)
|
|
}
|
|
|
|
override suspend fun getSidecar(ref: RemoteRef, kind: SidecarKind): ByteArray? =
|
|
withContext(Dispatchers.IO) {
|
|
val f = File(sidecarDir(ref), kind.fileName)
|
|
if (f.exists()) f.readBytes() else null
|
|
}
|
|
|
|
// ---- status ------------------------------------------------------------
|
|
|
|
override suspend fun storageLocationSummary(): StorageLocation =
|
|
withContext(Dispatchers.IO) {
|
|
ensureConnected()
|
|
val base = currentOrigin()
|
|
val host = runCatching { java.net.URI(base).host }.getOrNull() ?: base
|
|
val email = auth.load()?.email.orEmpty()
|
|
var count = 0
|
|
var bytes = 0L
|
|
var cursor: String? = null
|
|
do {
|
|
val page = list(cursor)
|
|
count += page.items.size
|
|
bytes += page.items.sumOf { it.ref.sizeBytes }
|
|
cursor = page.nextCursor
|
|
} while (cursor != null)
|
|
StorageLocation(
|
|
headline = "Custom SHONAR server at $host",
|
|
detail = if (email.isBlank()) "$count recordings synced."
|
|
else "Signed in as $email · $count recordings synced.",
|
|
syncedCount = count,
|
|
localOnlyCount = 0,
|
|
bytesUsed = bytes,
|
|
)
|
|
}
|
|
|
|
// ---- internals ---------------------------------------------------------
|
|
|
|
private fun ensureConnected() {
|
|
if (_authState.value != AuthState.CONNECTED) throw ProviderError.NotConnected()
|
|
}
|
|
|
|
private suspend fun currentOrigin(): String =
|
|
origin ?: auth.load()?.baseUrl ?: throw ProviderError.NotConnected()
|
|
|
|
private suspend fun login(base: String, email: String, password: String) {
|
|
val body = """{"email":${jsonStr(email)},"password":${jsonStr(password)},""" +
|
|
""""device_name":${jsonStr(deviceName)},"platform":${jsonStr(platform)}}"""
|
|
postUnauthed(base, "/api/v1/auth/login", body).use { resp ->
|
|
when (resp.code) {
|
|
200 -> {
|
|
persistSession(base, email, org.json.JSONObject(resp.body?.string().orEmpty()))
|
|
origin = base
|
|
_authState.value = AuthState.CONNECTED
|
|
}
|
|
401 -> throw ProviderError.Transient("Invalid email or password")
|
|
else -> throw ProviderError.Transient("Login failed (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
private suspend fun resumeWithTokens(credential: ProviderCredential.OAuthTokens) {
|
|
val base = origin ?: auth.load()?.baseUrl
|
|
?: throw ProviderError.InvalidUrl("No SHONAR server configured yet")
|
|
origin = base
|
|
auth.save(
|
|
ShonarSession(
|
|
baseUrl = base,
|
|
email = credential.accountLabel,
|
|
accessToken = credential.accessToken,
|
|
refreshToken = credential.refreshToken.orEmpty(),
|
|
expiresAtEpochSec = credential.expiresAtEpochSec,
|
|
deviceId = null,
|
|
)
|
|
)
|
|
executeAuthed(base) { token -> get(base, "/api/v1/auth/me", token) }.use { resp ->
|
|
if (resp.code != 200) throw ProviderError.AuthExpired()
|
|
}
|
|
_authState.value = AuthState.CONNECTED
|
|
}
|
|
|
|
private suspend fun persistSession(base: String, email: String, json: org.json.JSONObject) {
|
|
val nowSec = Instant.now().epochSecond
|
|
auth.save(
|
|
ShonarSession(
|
|
baseUrl = base,
|
|
email = email,
|
|
accessToken = json.getString("access_token"),
|
|
refreshToken = json.getString("refresh_token"),
|
|
expiresAtEpochSec = nowSec + json.optInt("expires_in", 900),
|
|
deviceId = json.optString("device_id", null).takeUnless { it.isNullOrBlank() },
|
|
)
|
|
)
|
|
}
|
|
|
|
/** Single-flight refresh; concurrent 401s must not double-refresh. */
|
|
private suspend fun refreshLocked(failedAccess: String): String = refreshMutex.withLock {
|
|
val current = auth.load() ?: throw ProviderError.NotConnected()
|
|
// A peer already refreshed while we queued — reuse its tokens.
|
|
if (current.accessToken != failedAccess && current.accessToken.isNotBlank()) {
|
|
return@withLock current.accessToken
|
|
}
|
|
if (current.refreshToken.isBlank()) {
|
|
_authState.value = AuthState.EXPIRED
|
|
throw ProviderError.AuthExpired()
|
|
}
|
|
val body = """{"refresh_token":${jsonStr(current.refreshToken)}}"""
|
|
try {
|
|
postUnauthed(current.baseUrl, "/api/v1/auth/refresh", body).use { resp ->
|
|
if (resp.code != 200) {
|
|
// Reuse detected or revoked family: stored tokens are dead.
|
|
auth.clearTokens()
|
|
_authState.value = AuthState.EXPIRED
|
|
throw ProviderError.AuthExpired()
|
|
}
|
|
persistSession(current.baseUrl, current.email, org.json.JSONObject(resp.body?.string().orEmpty()))
|
|
_authState.value = AuthState.CONNECTED
|
|
return@withLock auth.load()?.accessToken ?: throw ProviderError.AuthExpired()
|
|
}
|
|
} catch (e: ProviderError) {
|
|
throw e
|
|
} catch (e: Exception) {
|
|
throw ProviderError.Transient("Token refresh failed (${e.javaClass.simpleName})")
|
|
}
|
|
}
|
|
|
|
/** Execute an authenticated request; one transparent refresh+retry on 401. */
|
|
private suspend fun executeAuthed(
|
|
base: String,
|
|
build: (access: String) -> Request,
|
|
): Response = withContext(Dispatchers.IO) {
|
|
val session = auth.load() ?: throw ProviderError.NotConnected()
|
|
val first = client.newCall(build(session.accessToken)).execute()
|
|
if (first.code != 401) return@withContext first
|
|
first.close()
|
|
val fresh = refreshLocked(session.accessToken)
|
|
val retry = client.newCall(build(fresh)).execute()
|
|
if (retry.code == 401) {
|
|
retry.close()
|
|
_authState.value = AuthState.EXPIRED
|
|
throw ProviderError.AuthExpired()
|
|
}
|
|
retry
|
|
}
|
|
|
|
private fun get(base: String, path: String, access: String): Request =
|
|
Request.Builder().url(base + path).get()
|
|
.header("Authorization", "Bearer $access").build()
|
|
|
|
private fun post(base: String, path: String, access: String, json: String): Request =
|
|
Request.Builder().url(base + path)
|
|
.post(json.toRequestBody("application/json; charset=utf-8".toMediaType()))
|
|
.header("Authorization", "Bearer $access").build()
|
|
|
|
private fun put(base: String, path: String, access: String, json: String): Request =
|
|
Request.Builder().url(base + path)
|
|
.put(json.toRequestBody("application/json; charset=utf-8".toMediaType()))
|
|
.header("Authorization", "Bearer $access").build()
|
|
|
|
private fun patch(base: String, path: String, access: String, json: String): Request =
|
|
Request.Builder().url(base + path)
|
|
.patch(json.toRequestBody("application/json; charset=utf-8".toMediaType()))
|
|
.header("Authorization", "Bearer $access").build()
|
|
|
|
private fun postUnauthed(base: String, path: String, json: String): Response {
|
|
val req = Request.Builder().url(base + path)
|
|
.post(json.toRequestBody("application/json; charset=utf-8".toMediaType())).build()
|
|
return client.newCall(req).execute()
|
|
}
|
|
|
|
// Upload-session flow returns Triple(sessionId, chunkSize, declaredMime echo not needed).
|
|
private suspend fun createSession(base: String, draft: RecordingDraft): Pair<String, Int> {
|
|
val body = buildString {
|
|
append("""{"declared_mime_type":${jsonStr(draft.mime)},""")
|
|
append(""""declared_size_bytes":${draft.sizeBytes},""")
|
|
append(""""title":${jsonStr(draft.title)},""")
|
|
append(""""client_recording_id":${jsonStr(draft.id)}""")
|
|
if (draft.transcriptionModel != null) {
|
|
append(""","transcription_model":${jsonStr(draft.transcriptionModel)}""")
|
|
}
|
|
append("}")
|
|
}
|
|
executeAuthed(base) { token -> post(base, "/api/v1/uploads", token, body) }.use { resp ->
|
|
when (resp.code) {
|
|
201 -> {
|
|
val json = org.json.JSONObject(resp.body?.string().orEmpty())
|
|
return json.getString("id") to json.optInt("chunk_size_bytes", 16 * 1024 * 1024)
|
|
}
|
|
413 -> throw ProviderError.Transient("Recording exceeds the server size limit")
|
|
415 -> throw ProviderError.Transient("Audio type not accepted by the server")
|
|
else -> throw ProviderError.Transient("Upload rejected (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
}
|
|
|
|
private suspend fun uploadStatus(base: String, sessionId: String): Set<Int> {
|
|
executeAuthed(base) { token -> get(base, "/api/v1/uploads/$sessionId", token) }.use { resp ->
|
|
if (resp.code != 200) throw ProviderError.Transient("Upload status failed (HTTP ${resp.code})")
|
|
val arr = org.json.JSONObject(resp.body?.string().orEmpty())
|
|
.optJSONArray("received_chunk_indexes") ?: return emptySet()
|
|
return (0 until arr.length()).map { arr.getInt(it) }.toSet()
|
|
}
|
|
}
|
|
|
|
private suspend fun putChunk(base: String, sessionId: String, index: Int, bytes: ByteArray) {
|
|
val digest = MessageDigest.getInstance("SHA-256").digest(bytes).toHex()
|
|
val req = { token: String ->
|
|
Request.Builder().url("$base/api/v1/uploads/$sessionId/chunks/$index")
|
|
.put(bytes.toRequestBody("application/octet-stream".toMediaType()))
|
|
.header("Authorization", "Bearer $token")
|
|
.header("X-Chunk-Sha256", digest).build()
|
|
}
|
|
executeAuthed(base, req).use { resp ->
|
|
if (resp.code != 201) throw ProviderError.Transient("Chunk $index rejected (HTTP ${resp.code})")
|
|
}
|
|
}
|
|
|
|
private suspend fun finalize(base: String, sessionId: String, draft: RecordingDraft): String {
|
|
val recordedAt = Instant.ofEpochMilli(draft.createdAtEpochMs).toString()
|
|
val body = buildString {
|
|
append("""{"recorded_at":${jsonStr(recordedAt)},""")
|
|
append(""""duration_seconds":${draft.durationMs / 1000.0}""")
|
|
if (draft.transcriptionModel != null) {
|
|
append(""","transcription_model":${jsonStr(draft.transcriptionModel)}""")
|
|
}
|
|
append("}")
|
|
}
|
|
executeAuthed(base) { token -> post(base, "/api/v1/uploads/$sessionId/finalize", token, body) }
|
|
.use { resp ->
|
|
if (resp.code != 201) {
|
|
throw ProviderError.Transient("Upload finalize failed (HTTP ${resp.code})")
|
|
}
|
|
return org.json.JSONObject(resp.body?.string().orEmpty()).getString("id")
|
|
}
|
|
}
|
|
|
|
private fun parseRecording(json: org.json.JSONObject): RemoteRecording? {
|
|
if (!json.optBoolean("has_audio", false)) return null
|
|
val id = json.optString("id", "")
|
|
if (id.isBlank()) return null
|
|
val createdAt = runCatching {
|
|
Instant.parse(json.getString("recorded_at")).toEpochMilli()
|
|
}.getOrDefault(0L)
|
|
return RemoteRecording(
|
|
ref = RemoteRef(
|
|
providerId = ProviderRegistry.CUSTOM_SHONAR_ID,
|
|
key = id,
|
|
etag = null,
|
|
sizeBytes = json.optLong("size_bytes", 0),
|
|
),
|
|
title = json.optString("title", id),
|
|
createdAtEpochMs = createdAt,
|
|
durationMs = (json.optDouble("duration_seconds", 0.0) * 1000).toLong(),
|
|
mime = json.optString("mime_type", null) ?: "application/octet-stream",
|
|
)
|
|
}
|
|
|
|
companion object {
|
|
fun defaultClient(): OkHttpClient = OkHttpClient.Builder()
|
|
// No logging interceptor by policy (docs/server-providers.md §4):
|
|
// bodies would carry audio bytes and transcripts.
|
|
.connectTimeout(15, TimeUnit.SECONDS)
|
|
.readTimeout(60, TimeUnit.SECONDS)
|
|
.writeTimeout(60, TimeUnit.SECONDS)
|
|
.build()
|
|
|
|
/** JSON string literal with escaping; never pass secrets to message strings. */
|
|
internal fun jsonStr(raw: String): String = buildString {
|
|
append('"')
|
|
for (c in raw) when (c) {
|
|
'"' -> append("\\\"")
|
|
'\\' -> append("\\\\")
|
|
'\n' -> append("\\n")
|
|
'\r' -> append("\\r")
|
|
'\t' -> append("\\t")
|
|
else -> if (c < ' ') append("\\u%04x".format(c.code)) else append(c)
|
|
}
|
|
append('"')
|
|
}
|
|
|
|
private fun ByteArray.toHex(): String = joinToString("") { "%02x".format(it) }
|
|
|
|
private fun readSlice(file: File, offset: Long, max: Int): ByteArray {
|
|
file.inputStream().use { input ->
|
|
var skipped = 0L
|
|
while (skipped < offset) {
|
|
val n = input.skip(offset - skipped)
|
|
if (n <= 0) break
|
|
skipped += n
|
|
}
|
|
val buf = ByteArray(max)
|
|
var read = 0
|
|
while (read < max) {
|
|
val n = input.read(buf, read, max - read)
|
|
if (n < 0) break
|
|
read += n
|
|
}
|
|
return if (read == max) buf else buf.copyOf(read)
|
|
}
|
|
}
|
|
}
|
|
}
|