-
-
-
-
+ return
+
+
{t("providerConnections.title")}
{t("providerConnections.description")}
+
+
+
+
+
+
+
setToken(event.target.value)} />
+ {create.isError &&
}
+
+
+ {accounts.isLoading
+ ?
+ : accounts.isError
+ ?
void accounts.refetch()} />
+ : !accounts.data?.items.length
+ ?
+ : {accounts.data.items.map((account) =>
)}
+}
+
+function ProviderConnectionRow({ account }: { account: ProviderConnection }) {
+ const { formatError, t } = useI18n()
+ const queryClient = useQueryClient()
+ const [replacing, setReplacing] = useState(false)
+ const [token, setToken] = useState("")
+ const invalidate = () => void queryClient.invalidateQueries({ queryKey: ["provider-connections"] })
+ const remove = useMutation({ mutationFn: () => api(`/me/provider-connections/${account.id}`, { method: "DELETE" }), onSuccess: invalidate })
+ const test = useMutation({ mutationFn: () => api(`/me/provider-connections/${account.id}/test`, { method: "POST", body: "{}" }), onSuccess: invalidate })
+ const setDefault = useMutation({ mutationFn: () => api(`/me/provider-connections/${account.id}/default`, { method: "POST", body: "{}" }), onSuccess: invalidate })
+ const replace = useMutation({
+ mutationFn: () => api(`/me/provider-connections/${account.id}`, { method: "PATCH", body: JSON.stringify({ token }) }),
+ onSuccess: () => { setReplacing(false); setToken(""); invalidate() },
+ })
+ const title = account.displayName || account.login || account.providerType
+ const error = replace.error ?? remove.error ?? test.error ?? setDefault.error
+
+ return
+
+
+
+
{title}
{account.isDefault &&
{t("providerConnections.default")}}
+
{account.providerType} · {account.baseUrl}
+
{t("providerConnections.privateHint")}
- {create.isError &&
}
-
- }
-
+
+
+
+
+ {!account.isDefault && }
+
+
+
+ {test.isSuccess &&
{t("providerConnections.testSucceeded")}
}
+ {replacing &&
+ setToken(event.target.value)} />
+
+
}
+ {error &&
}
}
-function RepositoryList({ items, loading = false, compact = false, workspaceID }: { items: Repository[]; loading?: boolean; compact?: boolean; workspaceID?: string }) {
+function Repositories({ workspace, selectedRepositoryID, navigate }: { workspace: Workspace; selectedRepositoryID: string; navigate: (page: PageID, repositoryID?: string) => void }) {
const { t } = useI18n()
- const queryClient = useQueryClient(); const sync = useMutation({ mutationFn: ({ id, ref }: { id: string; ref: string }) => api(`/workspaces/${workspaceID}/repositories/${id}/sync`, { method: "POST", body: JSON.stringify({ ref }) }), onSuccess: () => void queryClient.invalidateQueries({ queryKey: ["repositories", workspaceID] }) })
- const cancel = useMutation({ mutationFn: (id: string) => api(`/workspaces/${workspaceID}/repositories/${id}/cancel-sync`, { method: "POST", body: "{}" }), onSuccess: () => void queryClient.invalidateQueries({ queryKey: ["repositories", workspaceID] }) })
- const remove = useMutation({ mutationFn: (id: string) => api(`/workspaces/${workspaceID}/repositories/${id}`, { method: "DELETE" }), onSuccess: () => void queryClient.invalidateQueries({ queryKey: ["repositories", workspaceID] }) })
- if (loading) return
- if (!items.length) return
- return
{items.map((item) =>
{item.cloneUrl}
{item.ref || t("repositories.remoteDefault")} · {item.currentCommitSha?.slice(0, 12) || t("repositories.notSynced")}
{item.lastErrorMessage &&
{item.lastErrorMessage}
}
{!compact && <>{item.state === "syncing" ?
:
}
}
)}
-}
+ const [adding, setAdding] = useState(false)
+ const [search, setSearch] = useState("")
+ const repositories = usePagedQuery
(["repositories", workspace.id], `/workspaces/${workspace.id}/repositories`, { refetchInterval: 4_000 })
+ const items = repositories.data?.items ?? []
+ const filteredItems = items.filter((repository) => `${repository.name} ${repository.remoteUrl} ${repository.ref}`.toLowerCase().includes(search.trim().toLowerCase()))
+ const selected = items.find((repository) => repository.id === selectedRepositoryID) ?? items[0]
-type ConnectionForm = { name: string; type: string; baseUrl: string; authType: string; username: string; token: string }
+ return
+
setAdding(true)}>{t("repositories.add")}}
+ />
+ {repositories.isLoading
+ ?
+ : repositories.isError
+ ? void repositories.refetch()} />
+ : !items.length
+ ? setAdding(true)}>{t("repositories.add")}} />
+ :
+
+ {selected &&
}
+
setAdding((value) => !value)}>{t("connections.new")}} />
- {adding && }
- {connections.isLoading ?
: !connections.data?.items.length ?
:
{connections.data.items.map((item) => )}
}
+
+ setRemoteUrl(event.target.value)} placeholder="https://github.com/owner/repository.git" />
+
+
setRepositoryRef(event.target.value)} placeholder={t("repositories.targetExample")} />
+
setName(event.target.value)} placeholder={t("repositories.namePlaceholder")} />
+
+
+ {create.isError &&
}
+
{t(item.secretConfigured ? "connections.credentialConfigured" : "connections.publicAccess")}{open &&
}
+ {items.map((item) =>
{item.ref} · {item.currentSnapshot?.commitSha.slice(0, 12) || t("repositories.notSynced")}
)}
+
{latestOperation && (latestOperation.status === "queued" || latestOperation.status === "running") ? : }
{item.currentSnapshot.commitSha} : t("repositories.notSynced")} mono />
+ {item.lastErrorMessage && {item.lastErrorMessage}
}
+ {mutationError && }
+
+ {tab === "activity" ? {operations.isLoading ? : operations.isError ? void operations.refetch()} />
: !operations.data?.items.length ?
: {operations.data.items.map((operation) => cancelOperation.mutate(operation.id)} />)}
} :
+
setRepositoryName(event.target.value)} />
+
+
+
+
+ {kind}{outcome && {outcome}}{formatDateTime(operation.createdAt)}{operation.resolvedCommitSha ? ` · ${operation.resolvedCommitSha.slice(0, 12)}` : ""}{operation.errorMessage && {operation.errorMessage}}
+ {active && }
+
+}
+
+type ChannelForm = { type: string; name: string; clientId: string; clientSecret: string; targetId: string; robotCode: string; isLark: boolean; senderAllowList: string; requireMention: boolean; prefix: string; notificationEvents: ChannelNotificationEvent[] }
+
+const channelNotificationOptions = [
+ { event: "analysis.started", label: "channels.notifyStarted" },
+ { event: "analysis.succeeded", label: "channels.notifySucceeded" },
+ { event: "analysis.failed", label: "channels.notifyFailed" },
+] as const
+
+const defaultChannelNotificationEvents: ChannelNotificationEvent[] = channelNotificationOptions.map(({ event }) => event)
function Channels({ workspace }: { workspace: Workspace }) {
const { formatError, t } = useI18n()
const queryClient = useQueryClient()
const [adding, setAdding] = useState(false)
- const { register, handleSubmit, reset, control } = useForm({ defaultValues: { type: "feishu", name: "", clientId: "", clientSecret: "", senderAllowList: "*", requireMention: true, prefix: "/moon" } })
- const channelType = useWatch({ control, name: "type" })
- const channels = useQuery({ queryKey: ["channels", workspace.id], queryFn: () => api<{ items: ChannelInstance[] }>(`/workspaces/${workspace.id}/channels`), refetchInterval: 5_000 })
+ const [form, setForm] = useState({ type: "feishu", name: "", clientId: "", clientSecret: "", targetId: "", robotCode: "", isLark: false, senderAllowList: "*", requireMention: true, prefix: "/moon", notificationEvents: defaultChannelNotificationEvents })
+ const channels = usePagedQuery(["channels", workspace.id], `/workspaces/${workspace.id}/channels`, { refetchInterval: 5_000 })
const create = useMutation({
- mutationFn: (form: ChannelForm) => api(`/workspaces/${workspace.id}/channels`, {
+ mutationFn: () => api(`/workspaces/${workspace.id}/channels`, {
method: "POST",
body: JSON.stringify({
type: form.type,
name: form.name,
- values: form.type === "feishu" ? { app_id: form.clientId, is_lark: false } : { client_id: form.clientId },
+ values: form.type === "feishu"
+ ? { app_id: form.clientId, is_lark: form.isLark, receive_id: form.targetId, receive_id_type: "chat_id" }
+ : { client_id: form.clientId, robot_code: form.robotCode, open_conversation_id: form.targetId },
secrets: form.type === "feishu" ? { app_secret: form.clientSecret } : { client_secret: form.clientSecret },
senderAllowList: form.senderAllowList.split(",").map((value) => value.trim()).filter(Boolean),
groupPolicy: { requireMention: form.requireMention, prefix: form.prefix },
+ notificationEvents: form.notificationEvents,
}),
}),
- onSuccess: () => { setAdding(false); reset(); void queryClient.invalidateQueries({ queryKey: ["channels", workspace.id] }) },
+ onSuccess: () => {
+ setAdding(false)
+ setForm({ type: "feishu", name: "", clientId: "", clientSecret: "", targetId: "", robotCode: "", isLark: false, senderAllowList: "*", requireMention: true, prefix: "/moon", notificationEvents: defaultChannelNotificationEvents })
+ void queryClient.invalidateQueries({ queryKey: ["channels", workspace.id] })
+ },
})
- return setAdding((value) => !value)}>{t("channels.add")}} />
- {adding && }
- {channels.isLoading ?
: !channels.data?.items.length ?
:
{channels.data.items.map((item) => )}
}
+
setAdding(true)}>{t("channels.add")}} />
+ setAdding(false)}>
+
+
setForm({ ...form, name: event.target.value })} placeholder={t("channels.engineeringBot")} />
+
setForm({ ...form, clientId: event.target.value })} autoComplete="off" />
+
setForm({ ...form, clientSecret: event.target.value })} autoComplete="new-password" />
+ {form.type === "dingtalk" &&
setForm({ ...form, robotCode: event.target.value })} autoComplete="off" />}
+
setForm({ ...form, targetId: event.target.value })} autoComplete="off" />
+ {form.type === "feishu" &&
}
+
setForm({ ...form, senderAllowList: event.target.value })} />{t("channels.allowedSendersHelp")}
+
setForm({ ...form, prefix: event.target.value })} />
+
+
+ {create.isError &&
}
+
+
+ {channels.isLoading
+ ?
+ : channels.isError
+ ?
void channels.refetch()} />
+ : !channels.data?.items.length
+ ? setAdding(true)}>{t("channels.add")}} />
+ : {channels.data.items.map((item) => )}
}
+
+
{item.type} · {t(item.secretConfigured ? "connections.credentialConfigured" : "channels.credentialMissing")}
{status.data?.lastErrorMessage &&
{status.data.lastErrorMessage}
}
{editing &&
}
+ {status.data?.lastErrorMessage && {status.data.lastErrorMessage}
}
+ {status.isError && {formatError(status.error)}
}
+ {toggle.isError && }
+ {item.enabled ? t("channels.receivingMessages") : t("channels.notReceivingMessages")}
+ setEditing(false)}>
+
}
function Messages({ workspace }: { workspace: Workspace }) {
const { t } = useI18n()
const [channelID, setChannelID] = useState("")
const [conversationID, setConversationID] = useState("")
- const channels = useQuery({ queryKey: ["channels", workspace.id], queryFn: () => api<{ items: ChannelInstance[] }>(`/workspaces/${workspace.id}/channels`) })
- const conversations = useQuery({ queryKey: ["conversations", workspace.id], queryFn: () => api<{ items: IMConversation[] }>(`/workspaces/${workspace.id}/conversations`) })
+ const channels = usePagedQuery(["channels", workspace.id], `/workspaces/${workspace.id}/channels`)
+ const conversations = usePagedQuery(["conversations", workspace.id], `/workspaces/${workspace.id}/conversations`)
const messages = useInfiniteQuery({
queryKey: ["messages", workspace.id, channelID, conversationID],
initialPageParam: "",
queryFn: ({ pageParam }) => {
- const query = new URLSearchParams({ pageSize: "30" })
+ const query = new URLSearchParams({ limit: "30" })
if (pageParam) query.set("cursor", pageParam)
if (channelID) query.set("channelId", channelID)
if (conversationID) query.set("conversationId", conversationID)
@@ -279,28 +1039,76 @@ function Messages({ workspace }: { workspace: Workspace }) {
})
const visibleConversations = (conversations.data?.items ?? []).filter((item) => !channelID || item.channelInstanceId === channelID)
const items = messages.data?.pages.flatMap((page) => page.items) ?? []
- return {messages.isLoading ?
: !items.length ?
: <>
{items.map((item) => )}
{messages.hasNextPage &&
}}
+
+
+
+
{messages.isLoading ?
: messages.isError ?
void messages.refetch()} /> : !items.length ? : {items.map((item) => )}
}{messages.hasNextPage && }
+
+
{item.senderDisplayName || item.senderCanonicalId}{t("messages.via", { channel: item.channelName })}
{item.content.content.text || `[${item.content.content.type}]`}
+ if (!compact) return {(item.senderDisplayName || item.senderCanonicalId).slice(0, 1).toUpperCase()}{item.senderDisplayName || item.senderCanonicalId}{t("messages.via", { channel: item.channelName })} · {formatDateTime(item.occurredAt)}
{item.content.content.text || `[${item.content.content.type}]`}
+
{item.senderDisplayName || item.senderCanonicalId}{t("messages.via", { channel: item.channelName })}
+
+
{item.content.content.text || `[${item.content.content.type}]`}
+
}
-function WorkspaceSettings({ workspace }: { workspace: Workspace }) {
- const { formatDateTime, formatError, formatRole, t } = useI18n()
+function WorkspaceSettings({ workspace, userID, initialTab, navigate }: { workspace: Workspace; userID: string; initialTab: SettingsTab; navigate: (page: PageID, repositoryID?: string, settingsTab?: SettingsTab) => void }) {
+ const { formatDateTime, formatError, t } = useI18n()
const queryClient = useQueryClient()
const [name, setName] = useState(workspace.name)
+ const [reportRetentionDays, setReportRetentionDays] = useState(workspace.reportRetentionDays)
const [newWorkspace, setNewWorkspace] = useState("")
const [inviteEmail, setInviteEmail] = useState("")
- const [inviteRole, setInviteRole] = useState<"admin" | "member" | "viewer">("member")
const [invitationLink, setInvitationLink] = useState("")
- const members = useQuery({ queryKey: ["workspace-members", workspace.id], queryFn: () => api<{ items: WorkspaceMember[] }>(`/workspaces/${workspace.id}/members`) })
- const invitations = useQuery({ queryKey: ["workspace-invitations", workspace.id], queryFn: () => api<{ items: WorkspaceInvitation[] }>(`/workspaces/${workspace.id}/invitations`) })
- const rename = useMutation({ mutationFn: () => api(`/workspaces/${workspace.id}`, { method: "PATCH", body: JSON.stringify({ name }) }), onSuccess: () => void queryClient.invalidateQueries({ queryKey: ["session"] }) })
- const create = useMutation({ mutationFn: () => api("/workspaces", { method: "POST", body: JSON.stringify({ name: newWorkspace }) }), onSuccess: () => { setNewWorkspace(""); void queryClient.invalidateQueries({ queryKey: ["session"] }) } })
+ const canManage = workspace.role === "owner" || workspace.role === "admin"
+ const members = usePagedQuery(["workspace-members", workspace.id], `/workspaces/${workspace.id}/members`)
+ const invitations = usePagedQuery(["workspace-invitations", workspace.id], `/workspaces/${workspace.id}/invitations`, { enabled: canManage })
+ const updateWorkspace = useMutation({
+ mutationFn: () => api(`/workspaces/${workspace.id}`, { method: "PATCH", body: JSON.stringify({ name, reportRetentionDays }) }),
+ onSuccess: () => void queryClient.invalidateQueries({ queryKey: ["session"] }),
+ })
+ const create = useMutation({
+ mutationFn: () => api("/workspaces", { method: "POST", body: JSON.stringify({ name: newWorkspace }) }),
+ onSuccess: () => { setNewWorkspace(""); void queryClient.invalidateQueries({ queryKey: ["session"] }) },
+ })
const invite = useMutation({
- mutationFn: () => api(`/workspaces/${workspace.id}/invitations`, { method: "POST", body: JSON.stringify({ email: inviteEmail, role: inviteRole }) }),
+ mutationFn: () => api(`/workspaces/${workspace.id}/invitations`, { method: "POST", body: JSON.stringify({ email: inviteEmail }) }),
onSuccess: (created) => {
const link = new URL("/app", window.location.origin)
link.hash = new URLSearchParams({ invitation: created.token ?? "" }).toString()
@@ -311,25 +1119,60 @@ function WorkspaceSettings({ workspace }: { workspace: Workspace }) {
})
const revokeInvitation = useMutation({ mutationFn: (id: string) => api(`/workspaces/${workspace.id}/invitations/${id}`, { method: "DELETE" }), onSuccess: () => void queryClient.invalidateQueries({ queryKey: ["workspace-invitations", workspace.id] }) })
const removeMember = useMutation({ mutationFn: (id: string) => api(`/workspaces/${workspace.id}/members/${id}`, { method: "DELETE", body: "{}" }), onSuccess: () => void queryClient.invalidateQueries({ queryKey: ["workspace-members", workspace.id] }) })
- return
-
+
+ return
+
} />
+
navigate("settings", undefined, tab)} ariaLabel={t("settings.tabs")} options={[{ value: "workspace", label: t("settings.workspaceTab") }, { value: "team", label: t("settings.teamTab") }, { value: "credentials", label: t("settings.credentialsTab") }]} />
+ {initialTab === "credentials" &&
}
+ {initialTab === "workspace" && <>
+
- {(workspace.role === "owner" || workspace.role === "admin") &&
+ }
+ {initialTab === "team" && <>
+ {canManage ?
{t("settings.invitations")}
{t("settings.invitationsDescription")}
- setInviteEmail(event.target.value)} placeholder="person@example.com" />
+ setInviteEmail(event.target.value)} placeholder="person@example.com" />
{invite.isError && }
- {invitationLink && }
- {invitations.isLoading ?
: invitations.data?.items.map((invitation) =>
{invitation.email}
{invitation.acceptedAt ? t("settings.accepted") : invitation.revokedAt ? t("settings.revoked") : new Date(invitation.expiresAt) <= new Date() ? t("settings.expired") : t("settings.expires", { date: formatDateTime(invitation.expiresAt) })}
{!invitation.acceptedAt && !invitation.revokedAt &&
}
)}
+ {invitations.isLoading ?
: invitations.isError ?
void invitations.refetch()} /> : !invitations.data?.items.length ? {t("settings.noInvitations")}
: invitations.data.items.map((invitation) =>
+
{invitation.email}
{invitation.acceptedAt ? t("settings.accepted") : invitation.revokedAt ? t("settings.revoked") : new Date(invitation.expiresAt) <= new Date() ? t("settings.expired") : t("settings.expires", { date: formatDateTime(invitation.expiresAt) })}
+ {!invitation.acceptedAt && !invitation.revokedAt &&
}
+
:
}
{t("settings.members")}
- {members.isLoading ?
: members.data?.items.map((member) => { const removable = member.userId !== workspace.createdBy && (workspace.role === "owner" || (workspace.role === "admin" && member.role !== "owner" && member.role !== "admin")); return
{(member.displayName || member.username || "U").slice(0, 1).toUpperCase()}
{member.displayName || member.username}
{member.email || member.userId}
{removable &&
}
})}
+ {members.isLoading ?
: members.isError ?
void members.refetch()} /> : !members.data?.items.length ? {t("settings.noMembers")}
: members.data.items.map((member) => {
+ const canRemove = canManage && member.role !== "owner" && member.userId !== userID && (workspace.role === "owner" || member.role === "member")
+ return
+
{(member.displayName || member.username || "U").slice(0, 1).toUpperCase()}
+
{member.displayName || member.username}
{member.email || member.userId}
+
{t(member.role === "owner" ? "settings.roleOwner" : member.role === "admin" ? "settings.roleAdmin" : "settings.roleMember")}
+ {canRemove &&
}
+
-
+ }
}
diff --git a/frontend/src/app/Authentication.tsx b/frontend/src/app/Authentication.tsx
index 4e08b0a..6d24f30 100644
--- a/frontend/src/app/Authentication.tsx
+++ b/frontend/src/app/Authentication.tsx
@@ -8,18 +8,23 @@ import { LanguageSwitcher } from "@/components/shared/LanguageSwitcher"
import { BrandMark, FormError } from "@/components/shared/presentation"
import { useI18n } from "@/i18n/context"
import { api, APIError } from "@/lib/api/client"
+import { logout, oauthRedirectURL } from "@/lib/auth"
import type { Session } from "@/lib/api/types"
export function AuthenticationFailure({ error, invitationToken }: { error: unknown; invitationToken: string }) {
const { formatError, t } = useI18n()
const failure = error instanceof APIError ? error : null
+ const login = useMutation({
+ mutationFn: oauthRedirectURL,
+ onSuccess: (redirectURL) => window.location.assign(redirectURL),
+ })
if (failure?.code === "account.suspended") {
return
}
if (failure?.code === "registration.closed") {
return
}
- if (failure?.status !== 401 || !failure.loginURL) {
+ if (failure?.status !== 401) {
return
}
return (
@@ -30,7 +35,8 @@ export function AuthenticationFailure({ error, invitationToken }: { error: unkno
? t("auth.invitation.signInDescription")
: t("auth.login.description")}
>
-
{title}
{description}
{action &&
{action}
}
{t("common.requestFailed")}
{formatError(error)}
{onRetry &&
}
{title}
{description &&
{description}
}
{action &&
{action}
}
+ {options.map((option) => )}
+
+}
+
+export function DefinitionItem({ label, value, mono = false, className }: { label: string; value: React.ReactNode; mono?: boolean; className?: string }) {
+ return {label}{value}
+}
+
+export function Drawer({ open, title, description, onClose, children, footer }: { open: boolean; title: string; description?: string; onClose: () => void; children: React.ReactNode; footer?: React.ReactNode }) {
+ const { t } = useI18n()
+ const dialogRef = useRef(null)
+ const onCloseRef = useRef(onClose)
+
+ useEffect(() => {
+ onCloseRef.current = onClose
+ }, [onClose])
+
+ useEffect(() => {
+ if (!open) return
+
+ const previousOverflow = document.body.style.overflow
+ const previousFocus = document.activeElement instanceof HTMLElement ? document.activeElement : null
+ const focusableSelector = 'a[href], button:not([disabled]), input:not([disabled]), select:not([disabled]), textarea:not([disabled]), [tabindex]:not([tabindex="-1"])'
+ const focusDialog = window.requestAnimationFrame(() => {
+ const firstFocusable = dialogRef.current?.querySelector(focusableSelector)
+ const focusTarget = firstFocusable ?? dialogRef.current
+ focusTarget?.focus()
+ })
+ const handleKeyDown = (event: KeyboardEvent) => {
+ if (event.key === "Escape") {
+ onCloseRef.current()
+ return
+ }
+ if (event.key !== "Tab" || !dialogRef.current) return
+
+ const focusable = Array.from(dialogRef.current.querySelectorAll(focusableSelector))
+ .filter((element) => element.getClientRects().length > 0)
+ if (!focusable.length) {
+ event.preventDefault()
+ dialogRef.current.focus()
+ return
+ }
+
+ const first = focusable[0]
+ const last = focusable[focusable.length - 1]
+ const active = document.activeElement
+ if (event.shiftKey && (active === first || !dialogRef.current.contains(active))) {
+ event.preventDefault()
+ last.focus()
+ } else if (!event.shiftKey && active === last) {
+ event.preventDefault()
+ first.focus()
+ }
+ }
+ document.body.style.overflow = "hidden"
+ window.addEventListener("keydown", handleKeyDown)
+
+ return () => {
+ window.cancelAnimationFrame(focusDialog)
+ document.body.style.overflow = previousOverflow
+ window.removeEventListener("keydown", handleKeyDown)
+ previousFocus?.focus()
+ }
+ }, [open])
+
+ if (!open) return null
+
+ return
+
+
+ {title}
{description &&
{description}
}
{children}
+ {footer && }
+
+
PreconditionFailedexists`))
- }))
- defer server.Close()
-
- store, err := New(strings.TrimPrefix(server.URL, "http://"), "access", "secret", WithRegion("us-east-1"))
- require.NoError(t, err)
- body := []byte("immutable")
- _, err = store.Put(context.Background(), storagecore.Location{Namespace: "artifacts", Key: "one.json"}, bytes.NewReader(body), int64(len(body)), storagecore.WithCreateOnly())
- require.Error(t, err)
- require.True(t, errors.Is(err, storagecore.ErrAlreadyExists), err)
-}
diff --git a/pkg/storage/storage.go b/pkg/storage/storage.go
deleted file mode 100644
index 4422885..0000000
--- a/pkg/storage/storage.go
+++ /dev/null
@@ -1,132 +0,0 @@
-// Package storage defines the portable binary storage contract used by MoonCode.
-// It deliberately contains no tenant, authorization, or database concepts.
-package storage
-
-import (
- "context"
- "errors"
- "fmt"
- "io"
- "path"
- "strings"
-)
-
-var (
- ErrNotFound = errors.New("storage: not found")
- ErrAlreadyExists = errors.New("storage: already exists")
- ErrInvalidKey = errors.New("storage: invalid key")
-)
-
-type Location struct {
- Namespace string
- Key string
-}
-
-func (l Location) Validate() error {
- if strings.TrimSpace(l.Namespace) == "" {
- return fmt.Errorf("%w: namespace is required", ErrInvalidKey)
- }
- if strings.ContainsAny(l.Namespace, "\\/\x00") || l.Namespace == "." || l.Namespace == ".." {
- return fmt.Errorf("%w: invalid namespace", ErrInvalidKey)
- }
- if l.Key == "" || strings.ContainsRune(l.Key, '\x00') || strings.ContainsRune(l.Key, '\\') {
- return fmt.Errorf("%w: invalid key", ErrInvalidKey)
- }
- if strings.HasPrefix(l.Key, "/") || path.Clean(l.Key) != l.Key || l.Key == "." || l.Key == ".." {
- return fmt.Errorf("%w: key must be a clean relative slash path", ErrInvalidKey)
- }
- for _, segment := range strings.Split(l.Key, "/") {
- if segment == "." || segment == ".." {
- return fmt.Errorf("%w: key contains a traversal segment", ErrInvalidKey)
- }
- }
- return nil
-}
-
-type Metadata struct {
- Location Location
- Size int64
- SHA256 string
- ETag string
- Version string
- ContentType string
-}
-
-type Store interface {
- Put(ctx context.Context, location Location, body io.Reader, size int64, opts ...PutOption) (Metadata, error)
- Open(ctx context.Context, location Location, opts ...OpenOption) (io.ReadCloser, error)
- Stat(ctx context.Context, location Location) (Metadata, error)
- Delete(ctx context.Context, location Location) error
-}
-
-type PutOptions struct {
- ContentType string
- SHA256 string
- CreateOnly bool
-}
-
-type PutOption func(*PutOptions)
-
-func WithContentType(contentType string) PutOption {
- return func(options *PutOptions) {
- options.ContentType = contentType
- }
-}
-
-func WithSHA256(checksum string) PutOption {
- return func(options *PutOptions) {
- options.SHA256 = checksum
- }
-}
-
-func WithCreateOnly() PutOption {
- return func(options *PutOptions) {
- options.CreateOnly = true
- }
-}
-
-func ResolvePutOptions(opts ...PutOption) PutOptions {
- var options PutOptions
- for _, option := range opts {
- if option != nil {
- option(&options)
- }
- }
- return options
-}
-
-type OpenOptions struct {
- Offset int64
- Length int64
-}
-
-type OpenOption func(*OpenOptions)
-
-func WithRange(offset, length int64) OpenOption {
- return func(options *OpenOptions) {
- options.Offset = offset
- options.Length = length
- }
-}
-
-func ResolveOpenOptions(opts ...OpenOption) OpenOptions {
- var options OpenOptions
- for _, option := range opts {
- if option != nil {
- option(&options)
- }
- }
- return options
-}
-
-type Wrapper func(Store) Store
-
-// Wrap applies wrappers in declaration order: the first wrapper is outermost.
-func Wrap(store Store, wrappers ...Wrapper) Store {
- for index := len(wrappers) - 1; index >= 0; index-- {
- if wrappers[index] != nil {
- store = wrappers[index](store)
- }
- }
- return store
-}
diff --git a/pkg/storage/storage_test.go b/pkg/storage/storage_test.go
deleted file mode 100644
index d6a621d..0000000
--- a/pkg/storage/storage_test.go
+++ /dev/null
@@ -1,73 +0,0 @@
-package storage
-
-import (
- "context"
- "io"
- "testing"
-
- "github.com/stretchr/testify/require"
-)
-
-func TestLocationValidate(t *testing.T) {
- tests := []struct {
- name string
- location Location
- wantErr bool
- }{
- {name: "valid", location: Location{Namespace: "artifacts", Key: "workspaces/one/report.json"}},
- {name: "empty namespace", location: Location{Key: "object"}, wantErr: true},
- {name: "absolute", location: Location{Namespace: "artifacts", Key: "/etc/passwd"}, wantErr: true},
- {name: "traversal", location: Location{Namespace: "artifacts", Key: "../passwd"}, wantErr: true},
- {name: "unclean", location: Location{Namespace: "artifacts", Key: "a//b"}, wantErr: true},
- {name: "backslash", location: Location{Namespace: "artifacts", Key: `a\b`}, wantErr: true},
- }
-
- for _, test := range tests {
- t.Run(test.name, func(t *testing.T) {
- err := test.location.Validate()
- if test.wantErr {
- require.ErrorIs(t, err, ErrInvalidKey)
- return
- }
- require.NoError(t, err)
- })
- }
-}
-
-func TestWrapUsesDeclarationOrder(t *testing.T) {
- var calls []string
- base := stubStore{onStat: func() { calls = append(calls, "base") }}
- wrap := func(name string) Wrapper {
- return func(next Store) Store {
- return stubStore{onStat: func() {
- calls = append(calls, name+":before")
- _, _ = next.Stat(context.Background(), Location{})
- calls = append(calls, name+":after")
- }}
- }
- }
-
- _, _ = Wrap(base, wrap("first"), wrap("second")).Stat(context.Background(), Location{})
- require.Equal(t, []string{"first:before", "second:before", "base", "second:after", "first:after"}, calls)
-}
-
-type stubStore struct {
- onStat func()
-}
-
-func (stubStore) Put(context.Context, Location, io.Reader, int64, ...PutOption) (Metadata, error) {
- return Metadata{}, nil
-}
-
-func (stubStore) Open(context.Context, Location, ...OpenOption) (io.ReadCloser, error) {
- return nil, nil
-}
-
-func (s stubStore) Stat(context.Context, Location) (Metadata, error) {
- if s.onStat != nil {
- s.onStat()
- }
- return Metadata{}, nil
-}
-
-func (stubStore) Delete(context.Context, Location) error { return nil }
diff --git a/pkg/storage/storagetest/contract.go b/pkg/storage/storagetest/contract.go
deleted file mode 100644
index 516348b..0000000
--- a/pkg/storage/storagetest/contract.go
+++ /dev/null
@@ -1,119 +0,0 @@
-// Package storagetest contains a reusable contract suite for storage.Store
-// adapters. Adapter integration tests can call Run against their real backend.
-package storagetest
-
-import (
- "bytes"
- "context"
- "errors"
- "fmt"
- "io"
- "sync"
- "testing"
- "time"
-
- "github.com/mooncode-ai/mooncode/pkg/storage"
-)
-
-func Run(t *testing.T, store storage.Store, namespace string) {
- t.Helper()
- prefix := fmt.Sprintf("contract/%d", time.Now().UnixNano())
-
- t.Run("put open stat range and delete", func(t *testing.T) {
- location := storage.Location{Namespace: namespace, Key: prefix + "/object.bin"}
- body := []byte("mooncode-storage-contract")
- metadata, err := store.Put(t.Context(), location, bytes.NewReader(body), int64(len(body)), storage.WithContentType("application/octet-stream"), storage.WithSHA256("fixture-checksum"), storage.WithCreateOnly())
- if err != nil {
- t.Fatalf("put: %v", err)
- }
- if metadata.Location != location {
- t.Fatalf("put location = %#v, want %#v", metadata.Location, location)
- }
- stat, err := store.Stat(t.Context(), location)
- if err != nil {
- t.Fatalf("stat: %v", err)
- }
- if stat.Size != int64(len(body)) || stat.SHA256 != "fixture-checksum" {
- t.Fatalf("stat metadata = %#v", stat)
- }
- reader, err := store.Open(t.Context(), location)
- if err != nil {
- t.Fatalf("open: %v", err)
- }
- contents, readErr := io.ReadAll(reader)
- closeErr := reader.Close()
- if readErr != nil || closeErr != nil || !bytes.Equal(contents, body) {
- t.Fatalf("read = %q, read error = %v, close error = %v", contents, readErr, closeErr)
- }
- rangeReader, err := store.Open(t.Context(), location, storage.WithRange(4, 4))
- if err != nil {
- t.Fatalf("open range: %v", err)
- }
- ranged, readErr := io.ReadAll(rangeReader)
- closeErr = rangeReader.Close()
- if readErr != nil || closeErr != nil || string(ranged) != "code" {
- t.Fatalf("range = %q, read error = %v, close error = %v", ranged, readErr, closeErr)
- }
- if _, err := store.Put(t.Context(), location, bytes.NewReader(body), int64(len(body)), storage.WithCreateOnly()); !errors.Is(err, storage.ErrAlreadyExists) {
- t.Fatalf("second create-only put error = %v, want ErrAlreadyExists", err)
- }
- if err := store.Delete(t.Context(), location); err != nil {
- t.Fatalf("delete: %v", err)
- }
- if err := store.Delete(t.Context(), location); err != nil {
- t.Fatalf("idempotent delete: %v", err)
- }
- if _, err := store.Stat(t.Context(), location); !errors.Is(err, storage.ErrNotFound) {
- t.Fatalf("stat deleted error = %v, want ErrNotFound", err)
- }
- })
-
- t.Run("atomic create only", func(t *testing.T) {
- location := storage.Location{Namespace: namespace, Key: prefix + "/atomic.bin"}
- start := make(chan struct{})
- errorsChannel := make(chan error, 2)
- var group sync.WaitGroup
- for _, value := range []string{"first", "second"} {
- group.Add(1)
- go func(value string) {
- defer group.Done()
- <-start
- _, err := store.Put(t.Context(), location, bytes.NewBufferString(value), int64(len(value)), storage.WithCreateOnly())
- errorsChannel <- err
- }(value)
- }
- close(start)
- group.Wait()
- close(errorsChannel)
- succeeded, alreadyExists := 0, 0
- for err := range errorsChannel {
- switch {
- case err == nil:
- succeeded++
- case errors.Is(err, storage.ErrAlreadyExists):
- alreadyExists++
- default:
- t.Fatalf("create-only error: %v", err)
- }
- }
- if succeeded != 1 || alreadyExists != 1 {
- t.Fatalf("successes = %d, already-exists = %d", succeeded, alreadyExists)
- }
- if err := store.Delete(t.Context(), location); err != nil {
- t.Fatalf("delete atomic fixture: %v", err)
- }
- })
-
- t.Run("validation and cancellation", func(t *testing.T) {
- invalid := storage.Location{Namespace: namespace, Key: "../escape"}
- if _, err := store.Stat(t.Context(), invalid); !errors.Is(err, storage.ErrInvalidKey) {
- t.Fatalf("invalid key error = %v, want ErrInvalidKey", err)
- }
- ctx, cancel := context.WithCancel(t.Context())
- cancel()
- location := storage.Location{Namespace: namespace, Key: prefix + "/cancelled.bin"}
- if _, err := store.Put(ctx, location, bytes.NewBufferString("cancelled"), 9); err == nil {
- t.Fatal("cancelled put unexpectedly succeeded")
- }
- })
-}
diff --git a/scripts/e2e/live-smoke.sh b/scripts/e2e/live-smoke.sh
new file mode 100755
index 0000000..eeabfcc
--- /dev/null
+++ b/scripts/e2e/live-smoke.sh
@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+base_url="${MOONCODE_E2E_BASE_URL:-}"
+session_cookie="${MOONCODE_E2E_SESSION_COOKIE:-}"
+
+if [[ -z "${base_url}" ]]; then
+ echo "MOONCODE_E2E_BASE_URL is required" >&2
+ exit 2
+fi
+if [[ "${base_url}" != http://* && "${base_url}" != https://* ]]; then
+ echo "MOONCODE_E2E_BASE_URL must use http or https" >&2
+ exit 2
+fi
+if [[ -z "${session_cookie}" ]]; then
+ echo "MOONCODE_E2E_SESSION_COOKIE is required" >&2
+ exit 2
+fi
+
+base_url="${base_url%/}"
+
+check_get() {
+ local path="$1"
+ local status
+
+ if ! status="$(curl \
+ --silent \
+ --show-error \
+ --request GET \
+ --connect-timeout 10 \
+ --max-time 30 \
+ --proto '=http,https' \
+ --header "Cookie: ${session_cookie}" \
+ --output /dev/null \
+ --write-out '%{http_code}' \
+ "${base_url}${path}")"; then
+ echo "GET ${path} failed" >&2
+ return 1
+ fi
+
+ if [[ "${status}" != 2?? ]]; then
+ echo "GET ${path} returned HTTP ${status}, expected 2xx" >&2
+ return 1
+ fi
+
+ echo "ok GET ${path} (${status})"
+}
+
+check_get "/healthz"
+check_get "/api/v1/session"
+check_get "/api/openapi.yaml"
diff --git a/sql/queries/analysis.sql b/sql/queries/analysis.sql
new file mode 100644
index 0000000..589b224
--- /dev/null
+++ b/sql/queries/analysis.sql
@@ -0,0 +1,139 @@
+-- name: CreateAnalysisRun :one
+INSERT INTO analysis_runs (
+ id, workspace_id, repository_id, snapshot_id, commit_sha, requested_by,
+ dimension_key, profile_id, profile_version, profile_snapshot, analyzer_version, idempotency_key, attempt, rerun_of, status
+)
+VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,'queued')
+ON CONFLICT (workspace_id, idempotency_key, attempt) DO NOTHING
+RETURNING *;
+
+-- name: LockAnalysisSnapshot :one
+SELECT snapshot.source_state
+FROM commit_snapshots AS snapshot
+JOIN repositories AS repository ON repository.id = snapshot.repository_id
+WHERE snapshot.id = $1
+ AND snapshot.repository_id = $2
+ AND repository.workspace_id = $3
+FOR UPDATE OF snapshot;
+
+-- name: GetAnalysisRunByAttempt :one
+SELECT * FROM analysis_runs
+WHERE workspace_id=$1 AND idempotency_key=$2 AND attempt=$3;
+
+-- name: GetAnalysisRun :one
+SELECT * FROM analysis_runs WHERE id=$1 AND workspace_id=$2;
+
+-- name: ListAnalysisRuns :many
+SELECT * FROM analysis_runs
+WHERE repository_id=sqlc.arg('repository_id')
+ AND workspace_id=sqlc.arg('workspace_id')
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (created_at, id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY created_at DESC, id DESC
+LIMIT sqlc.arg('limit');
+
+-- name: GetAnalysisRunWork :one
+SELECT * FROM analysis_runs WHERE id=$1;
+
+-- name: StartAnalysisRun :one
+UPDATE analysis_runs SET status='running', stage='authorize', started_at=now()
+WHERE id=$1 AND status='queued' RETURNING *;
+
+-- name: SetAnalysisRunStage :execrows
+UPDATE analysis_runs SET stage=$2
+WHERE id=$1 AND status='running';
+
+-- name: FinishAnalysisRun :one
+UPDATE analysis_runs
+SET status='succeeded', report_id=$2, failed_stage=NULL, error_code=NULL,
+ error_message=NULL, retryable=false, stage='complete', finished_at=$3
+WHERE id=$1 AND status='running' RETURNING *;
+
+-- name: FailAnalysisRun :one
+UPDATE analysis_runs
+SET status='failed', stage=$2, failed_stage=$2, error_code=$3, error_message=$4,
+ retryable=$5, finished_at=now()
+WHERE id=$1 AND status IN ('queued','running') RETURNING *;
+
+-- name: CancelAnalysisRun :one
+UPDATE analysis_runs SET status='cancelled', stage='cancelled', retryable=true, finished_at=now()
+WHERE id=$1 AND workspace_id=$2 AND status IN ('queued','running')
+RETURNING workflow_run_id;
+
+-- name: SetAnalysisWorkflowID :exec
+UPDATE analysis_runs SET workflow_run_id=$2 WHERE id=$1;
+
+-- name: CreateAnalysisReport :one
+INSERT INTO analysis_reports (
+ id, analysis_run_id, workspace_id, repository_id, snapshot_id, commit_sha,
+ source_ref, commit_author_name, commit_authored_at, commit_title,
+ dimension_key, profile_id, profile_version, profile_snapshot, analyzer_version, execution_environment,
+ started_at, finished_at, duration_ms, result, raw_artifact
+) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21)
+RETURNING *;
+
+-- name: AnalysisClock :one
+SELECT clock_timestamp()::timestamptz AS current_time;
+
+-- name: GetAnalysisReport :one
+SELECT * FROM analysis_reports WHERE id=$1 AND workspace_id=$2;
+
+-- name: GetAnalysisReportByRun :one
+SELECT * FROM analysis_reports WHERE analysis_run_id=$1;
+
+-- name: ListAnalysisProfiles :many
+SELECT p.*, v.id AS version_id, v.dimension_key, v.definition, v.created_at AS version_created_at
+FROM analysis_profiles p
+JOIN analysis_profile_versions v
+ ON v.workspace_id=p.workspace_id AND v.profile_id=p.id AND v.version=p.current_version
+WHERE p.workspace_id=sqlc.arg('workspace_id') AND p.archived_at IS NULL
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (p.created_at, p.id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY p.created_at DESC, p.id DESC
+LIMIT sqlc.arg('limit');
+
+-- name: GetAnalysisProfile :one
+SELECT p.*, v.id AS version_id, v.dimension_key, v.definition, v.created_at AS version_created_at
+FROM analysis_profiles p
+JOIN analysis_profile_versions v
+ ON v.workspace_id=p.workspace_id AND v.profile_id=p.id AND v.version=p.current_version
+WHERE p.id=$1 AND p.workspace_id=$2 AND p.archived_at IS NULL;
+
+-- name: GetDefaultAnalysisProfile :one
+SELECT p.*, v.id AS version_id, v.dimension_key, v.definition, v.created_at AS version_created_at
+FROM analysis_profiles p
+JOIN analysis_profile_versions v
+ ON v.workspace_id=p.workspace_id AND v.profile_id=p.id AND v.version=p.current_version
+WHERE p.workspace_id=$1 AND p.archived_at IS NULL
+ORDER BY p.created_at, p.id
+LIMIT 1;
+
+-- name: CreateAnalysisProfile :one
+INSERT INTO analysis_profiles (id, workspace_id, name, current_version, created_by)
+VALUES ($1,$2,$3,1,$4) RETURNING *;
+
+-- name: CreateAnalysisProfileVersion :one
+INSERT INTO analysis_profile_versions (id, workspace_id, profile_id, version, dimension_key, definition, created_by)
+VALUES ($1,$2,$3,$4,$5,$6,$7) RETURNING *;
+
+-- name: AdvanceAnalysisProfile :one
+UPDATE analysis_profiles
+SET name=$3, current_version=current_version+1, updated_at=now()
+WHERE id=$1 AND workspace_id=$2 AND current_version=$4 AND archived_at IS NULL
+RETURNING *;
+
+-- name: ArchiveAnalysisProfile :execrows
+UPDATE analysis_profiles SET archived_at=now(), updated_at=now()
+WHERE id=$1 AND workspace_id=$2 AND archived_at IS NULL;
+
+-- name: LockActiveAnalysisProfiles :many
+SELECT id FROM analysis_profiles
+WHERE workspace_id=$1 AND archived_at IS NULL
+ORDER BY id
+FOR UPDATE;
+-- name: CountActiveAnalysisRuns :one
+SELECT count(*) FROM analysis_runs WHERE workspace_id=$1 AND status IN ('queued','running');
diff --git a/sql/queries/audit.sql b/sql/queries/audit.sql
new file mode 100644
index 0000000..7fba724
--- /dev/null
+++ b/sql/queries/audit.sql
@@ -0,0 +1,18 @@
+-- name: CreateAuditEvent :exec
+INSERT INTO audit_events (
+ id,
+ workspace_id,
+ actor_user_id,
+ action,
+ resource_type,
+ resource_id,
+ metadata
+) VALUES (
+ sqlc.arg(id),
+ sqlc.narg(workspace_id),
+ sqlc.narg(actor_user_id),
+ sqlc.arg(action),
+ sqlc.arg(resource_type),
+ sqlc.narg(resource_id),
+ sqlc.arg(metadata)
+);
diff --git a/sql/queries/channel_commands.sql b/sql/queries/channel_commands.sql
new file mode 100644
index 0000000..fdb0899
--- /dev/null
+++ b/sql/queries/channel_commands.sql
@@ -0,0 +1,68 @@
+-- name: GetChannelExternalIdentity :one
+SELECT user_id FROM channel_external_identities
+WHERE channel_id=$1 AND sender_canonical_id=$2;
+
+-- name: CheckActiveChannelVersion :one
+SELECT config_version FROM channels
+WHERE id=$1 AND workspace_id=$2 AND enabled AND config_version=$3;
+
+-- name: CreateChannelIdentityLink :one
+INSERT INTO channel_identity_links (
+ id, workspace_id, channel_id, channel_version, sender_canonical_id, token_hash, expires_at
+)
+SELECT sqlc.arg('id'), sqlc.arg('workspace_id'), sqlc.arg('channel_id'),
+ sqlc.arg('channel_version'), sqlc.arg('sender_canonical_id'),
+ sqlc.arg('token_hash'), sqlc.arg('expires_at')
+FROM channels
+WHERE id=sqlc.arg('channel_id')
+ AND workspace_id=sqlc.arg('workspace_id')
+ AND enabled
+ AND config_version=sqlc.arg('channel_version')
+RETURNING *;
+
+-- name: GetChannelIdentityLink :one
+SELECT l.* FROM channel_identity_links l
+JOIN channels c ON c.id=l.channel_id
+WHERE l.token_hash=$1
+ AND l.consumed_at IS NULL
+ AND l.expires_at > now()
+ AND c.enabled
+ AND c.config_version=l.channel_version;
+
+-- name: ConsumeChannelIdentityLink :execrows
+UPDATE channel_identity_links l SET consumed_at=now()
+FROM channels c
+WHERE l.id=$1
+ AND l.consumed_at IS NULL
+ AND l.expires_at > now()
+ AND c.id=l.channel_id
+ AND c.enabled
+ AND c.config_version=l.channel_version;
+
+-- name: UpsertChannelExternalIdentity :exec
+INSERT INTO channel_external_identities (
+ id, workspace_id, channel_id, sender_canonical_id, user_id
+) VALUES ($1,$2,$3,$4,$5)
+ON CONFLICT (channel_id, sender_canonical_id) DO UPDATE
+SET user_id=excluded.user_id;
+
+-- name: GetConversationIDByExternal :one
+SELECT conversation.id FROM conversations AS conversation
+JOIN channels AS channel ON channel.id=conversation.channel_id
+WHERE conversation.channel_id=sqlc.arg('channel_id')
+ AND conversation.external_id=sqlc.arg('external_id')
+ AND channel.workspace_id=sqlc.arg('workspace_id')
+ AND channel.enabled
+ AND channel.config_version=sqlc.arg('channel_version');
+
+-- name: GetConversationBinding :one
+SELECT repository_id FROM conversation_bindings WHERE conversation_id=$1;
+
+-- name: UpsertConversationBinding :exec
+INSERT INTO conversation_bindings (conversation_id, workspace_id, repository_id, bound_by)
+VALUES ($1,$2,$3,$4)
+ON CONFLICT (conversation_id) DO UPDATE
+SET repository_id=excluded.repository_id, bound_by=excluded.bound_by, updated_at=now();
+
+-- name: DeleteConversationBinding :execrows
+DELETE FROM conversation_bindings WHERE conversation_id=$1;
diff --git a/sql/queries/channels.sql b/sql/queries/channels.sql
index fd7b587..92c13c4 100644
--- a/sql/queries/channels.sql
+++ b/sql/queries/channels.sql
@@ -1,142 +1,124 @@
--- name: CreateChannelInstance :one
-INSERT INTO channel_instances (id, workspace_id, type, name, enabled, config, secret_ref)
-VALUES (sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(type), sqlc.arg(name), false,
- sqlc.arg(config), sqlc.narg(secret_ref))
-RETURNING *;
-
--- name: ListChannelInstances :many
-SELECT * FROM channel_instances
-WHERE workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL
-ORDER BY created_at DESC, id DESC;
-
--- name: GetChannelInstance :one
-SELECT * FROM channel_instances
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL;
-
--- name: ListEnabledChannelInstances :many
-SELECT * FROM channel_instances
-WHERE enabled = true AND deleted_at IS NULL
-ORDER BY id;
+-- name: ListChannels :many
+SELECT * FROM channels
+WHERE workspace_id=sqlc.arg('workspace_id')
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (created_at, id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY created_at DESC, id DESC
+LIMIT sqlc.arg('limit');
+
+-- name: CreateChannel :one
+INSERT INTO channels (id, workspace_id, type, name, enabled, secret_ciphertext, secret_nonce, key_version, config)
+VALUES ($1,$2,$3,$4,false,$5,$6,$7,$8) RETURNING *;
+
+-- name: GetChannel :one
+SELECT * FROM channels WHERE id=$1 AND workspace_id=$2;
+
+-- name: ListChannelSubscriptions :many
+SELECT channel_id, event_type FROM channel_subscriptions
+WHERE workspace_id=$1
+ORDER BY channel_id, event_type;
+
+-- name: ListSubscribedChannels :many
+SELECT c.* FROM channels c
+JOIN channel_subscriptions subscription ON subscription.channel_id=c.id AND subscription.workspace_id=c.workspace_id
+WHERE c.workspace_id=$1 AND c.enabled AND subscription.event_type=$2
+ORDER BY c.created_at;
+
+-- name: DeleteChannelSubscriptions :exec
+DELETE FROM channel_subscriptions WHERE workspace_id=$1 AND channel_id=$2;
+
+-- name: CreateChannelSubscription :exec
+INSERT INTO channel_subscriptions (id, workspace_id, channel_id, event_type)
+VALUES ($1,$2,$3,$4);
+
+-- name: ListRuntimeChannels :many
+SELECT * FROM channels
+WHERE enabled AND secret_ciphertext IS NOT NULL AND secret_nonce IS NOT NULL AND key_version IS NOT NULL
+ORDER BY created_at;
+
+-- name: LockChannelRuntime :one
+SELECT config_version FROM channels
+WHERE id=$1 AND workspace_id=$2 AND enabled
+FOR KEY SHARE;
+
+-- name: MarkChannelRuntimeConnected :execrows
+UPDATE channels SET runtime_status='connected', last_connected_at=now(), last_error_message=NULL, updated_at=now()
+WHERE id=$1 AND enabled AND config_version=$2;
+
+-- name: MarkChannelRuntimeError :execrows
+UPDATE channels SET runtime_status='error', last_error_message=$3, updated_at=now()
+WHERE id=$1 AND enabled AND config_version=$2;
+
+-- name: UpdateChannel :one
+UPDATE channels SET name=$3, config=$4, config_version=config_version+1,
+ runtime_status=CASE WHEN enabled THEN 'starting' ELSE 'disabled' END,
+ last_error_message=NULL, updated_at=now()
+WHERE id=$1 AND workspace_id=$2 AND config_version=$5 RETURNING *;
+
+-- name: RotateChannelCredential :one
+UPDATE channels SET secret_ciphertext=$3, secret_nonce=$4, key_version=$5,
+ config_version=config_version+1,
+ runtime_status=CASE WHEN enabled THEN 'starting' ELSE 'disabled' END,
+ last_error_message=NULL, updated_at=now()
+WHERE id=$1 AND workspace_id=$2 AND config_version=$6 RETURNING *;
-- name: SetChannelEnabled :one
-UPDATE channel_instances SET enabled = sqlc.arg(enabled), config_version = config_version + 1,
- updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL
-RETURNING *;
-
--- name: SoftDeleteChannel :one
-UPDATE channel_instances
-SET enabled = false, secret_ref = NULL, deleted_at = now(), config_version = config_version + 1, updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL
-RETURNING *;
-
--- name: UpdateChannelInstance :one
-UPDATE channel_instances SET name = sqlc.arg(name), config = sqlc.arg(config),
- secret_ref = sqlc.narg(secret_ref), config_version = config_version + 1, updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id)
- AND config_version = sqlc.arg(expected_version) AND deleted_at IS NULL
-RETURNING *;
-
--- name: AcquireChannelLease :one
-INSERT INTO channel_leases (channel_instance_id, owner, lease_until, fencing_token)
-VALUES (sqlc.arg(channel_instance_id), sqlc.arg(owner), sqlc.arg(lease_until), 1)
-ON CONFLICT (channel_instance_id) DO UPDATE
-SET owner = EXCLUDED.owner, lease_until = EXCLUDED.lease_until,
- fencing_token = CASE WHEN channel_leases.owner = EXCLUDED.owner
- THEN channel_leases.fencing_token ELSE channel_leases.fencing_token + 1 END,
- updated_at = now()
-WHERE channel_leases.owner = EXCLUDED.owner OR channel_leases.lease_until < now()
-RETURNING *;
-
--- name: RenewChannelLease :execrows
-UPDATE channel_leases SET lease_until = sqlc.arg(lease_until), updated_at = now()
-WHERE channel_instance_id = sqlc.arg(channel_instance_id) AND owner = sqlc.arg(owner)
- AND fencing_token = sqlc.arg(fencing_token);
-
--- name: ReleaseChannelLease :exec
-DELETE FROM channel_leases
-WHERE channel_instance_id = sqlc.arg(channel_instance_id) AND owner = sqlc.arg(owner)
- AND fencing_token = sqlc.arg(fencing_token);
-
--- name: SetChannelRuntimeStatus :execrows
-INSERT INTO channel_runtime_status (
- channel_instance_id, state, backend_instance_id, fencing_token,
- last_connected_at, last_error_code, last_error_message
-) SELECT
- lease.channel_instance_id, sqlc.arg(state), sqlc.arg(backend_instance_id),
- sqlc.arg(fencing_token), sqlc.narg(last_connected_at), sqlc.arg(last_error_code),
- sqlc.arg(last_error_message)
-FROM channel_leases AS lease
-WHERE lease.channel_instance_id = sqlc.arg(channel_instance_id)
- AND lease.owner = sqlc.arg(backend_instance_id)
- AND lease.fencing_token = sqlc.arg(fencing_token)
- AND lease.lease_until > now()
-ON CONFLICT (channel_instance_id) DO UPDATE SET
- state = EXCLUDED.state, backend_instance_id = EXCLUDED.backend_instance_id,
- fencing_token = EXCLUDED.fencing_token,
- last_connected_at = COALESCE(EXCLUDED.last_connected_at, channel_runtime_status.last_connected_at),
- last_error_code = EXCLUDED.last_error_code,
- last_error_message = EXCLUDED.last_error_message, updated_at = now()
-WHERE channel_runtime_status.fencing_token <= EXCLUDED.fencing_token;
-
--- name: GetChannelRuntimeStatus :one
-SELECT * FROM channel_runtime_status WHERE channel_instance_id = sqlc.arg(channel_instance_id);
-
--- name: InsertInboxEvent :one
-INSERT INTO inbox_events (channel_instance_id, external_event_id, payload_hash)
-VALUES (sqlc.arg(channel_instance_id), sqlc.arg(external_event_id), sqlc.arg(payload_hash))
-ON CONFLICT DO NOTHING
-RETURNING external_event_id;
-
--- name: UpsertIMConversation :one
-INSERT INTO im_conversations (id, workspace_id, channel_instance_id, external_id, type, title)
-VALUES (sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(channel_instance_id),
- sqlc.arg(external_id), sqlc.arg(type), sqlc.arg(title))
-ON CONFLICT (channel_instance_id, external_id) DO UPDATE
-SET type = EXCLUDED.type, title = CASE WHEN EXCLUDED.title = '' THEN im_conversations.title ELSE EXCLUDED.title END,
- updated_at = now()
-RETURNING *;
-
--- name: UpsertIMSender :one
-INSERT INTO im_senders (id, workspace_id, channel_type, canonical_id, display_name)
-VALUES (sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(channel_type),
- sqlc.arg(canonical_id), sqlc.arg(display_name))
-ON CONFLICT (workspace_id, canonical_id) DO UPDATE
-SET display_name = CASE WHEN EXCLUDED.display_name = '' THEN im_senders.display_name ELSE EXCLUDED.display_name END,
- updated_at = now()
-RETURNING *;
-
--- name: InsertIMMessage :one
-INSERT INTO im_messages (
- id, workspace_id, channel_instance_id, conversation_id, sender_id,
- external_message_id, content, occurred_at
-) VALUES (
- sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(channel_instance_id),
- sqlc.arg(conversation_id), sqlc.arg(sender_id), sqlc.arg(external_message_id),
- sqlc.arg(content), sqlc.arg(occurred_at)
+UPDATE channels SET enabled=$3, runtime_status=CASE WHEN $3 THEN 'starting' ELSE 'disabled' END,
+ last_error_message=CASE WHEN $3 THEN NULL ELSE last_error_message END, updated_at=now()
+WHERE id=$1 AND workspace_id=$2 RETURNING *;
+
+-- name: DeleteChannel :execrows
+DELETE FROM channels WHERE id=$1 AND workspace_id=$2;
+
+-- name: UpsertConversation :one
+INSERT INTO conversations (id, workspace_id, channel_id, external_id, type, title)
+VALUES ($1,$2,$3,$4,$5,$6)
+ON CONFLICT (channel_id, external_id) DO UPDATE
+SET type=excluded.type, title=excluded.title
+RETURNING id;
+
+-- name: InsertInboundMessage :execrows
+INSERT INTO messages (
+ id, workspace_id, channel_id, conversation_id, external_id,
+ sender_canonical_id, sender_display_name, content, occurred_at
)
-ON CONFLICT (channel_instance_id, external_message_id) DO NOTHING
-RETURNING *;
-
--- name: ListIMMessages :many
+VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9)
+ON CONFLICT (channel_id, external_id) DO NOTHING;
+
+-- name: ListConversations :many
+SELECT c.id, c.workspace_id, c.channel_id, c.external_id, c.type, c.title, c.created_at,
+ ch.name AS channel_name, ch.type AS channel_type
+FROM conversations c JOIN channels ch ON ch.id=c.channel_id
+WHERE c.workspace_id=sqlc.arg('workspace_id')
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (c.created_at, c.id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY c.created_at DESC, c.id DESC
+LIMIT sqlc.arg('limit');
+
+-- name: ListMessages :many
SELECT m.*, c.external_id AS conversation_external_id, c.type AS conversation_type,
- s.canonical_id AS sender_canonical_id, s.display_name AS sender_display_name,
- ci.name AS channel_name, ci.type AS channel_type
-FROM im_messages m
-JOIN im_conversations c ON c.id = m.conversation_id AND c.workspace_id = m.workspace_id
-JOIN im_senders s ON s.id = m.sender_id AND s.workspace_id = m.workspace_id
-JOIN channel_instances ci ON ci.id = m.channel_instance_id AND ci.workspace_id = m.workspace_id
-WHERE m.workspace_id = sqlc.arg(workspace_id)
- AND (sqlc.narg(channel_instance_id)::uuid IS NULL OR m.channel_instance_id = sqlc.narg(channel_instance_id))
- AND (sqlc.narg(conversation_id)::uuid IS NULL OR m.conversation_id = sqlc.narg(conversation_id))
- AND (sqlc.narg(before_time)::timestamptz IS NULL OR (m.occurred_at, m.id) < (sqlc.narg(before_time), sqlc.narg(before_id)::uuid))
+ ch.name AS channel_name, ch.type AS channel_type
+FROM messages m
+JOIN conversations c ON c.id=m.conversation_id
+JOIN channels ch ON ch.id=m.channel_id
+WHERE m.workspace_id=sqlc.arg('workspace_id')
+ AND (sqlc.narg('channel_id')::uuid IS NULL OR m.channel_id=sqlc.narg('channel_id'))
+ AND (sqlc.narg('conversation_id')::uuid IS NULL OR m.conversation_id=sqlc.narg('conversation_id'))
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (m.occurred_at, m.id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
ORDER BY m.occurred_at DESC, m.id DESC
-LIMIT sqlc.arg(page_size);
-
--- name: ListIMConversations :many
-SELECT c.*, ci.name AS channel_name, ci.type AS channel_type
-FROM im_conversations c
-JOIN channel_instances ci ON ci.id = c.channel_instance_id AND ci.workspace_id = c.workspace_id
-WHERE c.workspace_id = sqlc.arg(workspace_id)
-ORDER BY c.updated_at DESC, c.id DESC
-LIMIT sqlc.arg(page_size);
+LIMIT sqlc.arg('limit');
+
+-- name: GetWorkspaceOverview :one
+SELECT
+ (SELECT count(*) FROM repositories rr WHERE rr.workspace_id=$1 AND rr.status <> 'deleted') AS repository_count,
+ (SELECT count(*) FROM repositories rr WHERE rr.workspace_id=$1 AND rr.status <> 'deleted' AND rr.current_snapshot_id IS NULL) AS repository_without_code_count,
+ (SELECT count(*) FROM repository_operations o JOIN repositories r ON r.id=o.repository_id WHERE r.workspace_id=$1 AND o.status='failed' AND o.created_at > now()-interval '7 days') AS recent_failed_sync_count,
+ (SELECT count(*) FROM analysis_runs ar WHERE ar.workspace_id=$1 AND ar.status IN ('queued','running')) AS active_analysis_count,
+ (SELECT count(*) FROM channels cc WHERE cc.workspace_id=$1 AND cc.enabled) AS active_channel_count;
diff --git a/sql/queries/identity.sql b/sql/queries/identity.sql
index 4ada781..a895a99 100644
--- a/sql/queries/identity.sql
+++ b/sql/queries/identity.sql
@@ -1,132 +1,103 @@
--- name: UpsertUser :one
-INSERT INTO users (
- id, issuer, external_subject, username, email, display_name, last_seen_at
-) VALUES (
- sqlc.arg(id), sqlc.arg(issuer), sqlc.arg(external_subject),
- sqlc.arg(username), sqlc.arg(email), sqlc.arg(display_name), now()
-)
-ON CONFLICT (issuer, external_subject) DO UPDATE SET
- username = EXCLUDED.username,
- email = EXCLUDED.email,
- display_name = EXCLUDED.display_name,
- last_seen_at = now(),
- updated_at = now()
+-- name: GetUserByIdentity :one
+SELECT u.* FROM users u
+JOIN oauth_identities oi ON oi.user_id = u.id
+WHERE oi.issuer = $1 AND oi.subject = $2;
+
+-- name: CreateUser :one
+INSERT INTO users (id, display_name, username, email, status)
+VALUES ($1, $2, $3, $4, $5)
RETURNING *;
--- name: GetUserByExternalIdentity :one
-SELECT * FROM users
-WHERE issuer = sqlc.arg(issuer) AND external_subject = sqlc.arg(external_subject);
+-- name: CreateOAuthIdentity :exec
+INSERT INTO oauth_identities (issuer, subject, user_id)
+VALUES ($1, $2, $3);
--- name: GetUserForUpdate :one
-SELECT * FROM users
-WHERE id = sqlc.arg(id)
-FOR UPDATE;
+-- name: UpdateUserProfile :one
+UPDATE users SET display_name = $2, username = $3, email = $4, updated_at = now()
+WHERE id = $1 RETURNING *;
-- name: ActivateUser :one
-UPDATE users SET
- status = 'active',
- terms_version = sqlc.arg(terms_version),
- privacy_version = sqlc.arg(privacy_version),
- activated_at = COALESCE(activated_at, now()),
- agreements_accepted_at = now(),
- updated_at = now()
-WHERE id = sqlc.arg(id) AND status IN ('pending', 'active')
+UPDATE users
+SET status = 'active', accepted_terms_at = now(), accepted_privacy_at = now(), updated_at = now()
+WHERE id = $1 AND status = 'pending'
RETURNING *;
+-- name: GetUser :one
+SELECT * FROM users WHERE id = $1;
+
-- name: CreateWorkspace :one
-INSERT INTO workspaces (id, name, slug, kind, created_by)
-VALUES (sqlc.arg(id), sqlc.arg(name), sqlc.arg(slug), sqlc.arg(kind), sqlc.arg(created_by))
+INSERT INTO workspaces (id, name, slug, created_by)
+VALUES ($1, $2, $3, $4)
RETURNING *;
--- name: AddWorkspaceMember :one
+-- name: CreateRetentionCleanupSchedule :exec
+INSERT INTO retention_cleanup_schedules (workspace_id)
+VALUES ($1);
+
+-- name: AddWorkspaceMember :exec
INSERT INTO workspace_members (workspace_id, user_id, role)
-VALUES (sqlc.arg(workspace_id), sqlc.arg(user_id), sqlc.arg(role))
-ON CONFLICT (workspace_id, user_id) DO NOTHING
-RETURNING *;
+VALUES ($1, $2, $3)
+ON CONFLICT (workspace_id, user_id) DO NOTHING;
--- name: GetPersonalWorkspaceForUser :one
-SELECT w.*
-FROM workspaces w
-JOIN workspace_members wm ON wm.workspace_id = w.id
-WHERE wm.user_id = sqlc.arg(user_id)
- AND w.kind = 'personal'
- AND w.deleted_at IS NULL
-ORDER BY w.created_at
-LIMIT 1;
-
--- name: ListWorkspacesForUser :many
-SELECT w.id, w.name, w.slug, w.kind, w.created_by, w.created_at, w.updated_at, wm.role
-FROM workspaces w
+-- name: GetWorkspaceMembership :one
+SELECT w.*, wm.role FROM workspaces w
JOIN workspace_members wm ON wm.workspace_id = w.id
-WHERE wm.user_id = sqlc.arg(user_id)
- AND w.deleted_at IS NULL
-ORDER BY w.created_at, w.id;
+JOIN users u ON u.id = wm.user_id AND u.status = 'active'
+WHERE w.id = $1 AND wm.user_id = $2;
--- name: GetWorkspaceMembership :one
-SELECT w.id, w.name, w.slug, w.kind, w.created_by, w.created_at, w.updated_at, wm.role
-FROM workspaces w
+-- name: ListUserWorkspaces :many
+SELECT w.*, wm.role FROM workspaces w
JOIN workspace_members wm ON wm.workspace_id = w.id
-WHERE w.id = sqlc.arg(workspace_id)
- AND wm.user_id = sqlc.arg(user_id)
- AND w.deleted_at IS NULL;
+WHERE wm.user_id = $1
+ORDER BY w.created_at;
-- name: UpdateWorkspace :one
-UPDATE workspaces SET name = sqlc.arg(name), updated_at = now()
-WHERE id = sqlc.arg(id) AND deleted_at IS NULL
+UPDATE workspaces
+SET name = COALESCE(sqlc.narg('name')::text, name),
+ report_retention_days = COALESCE(sqlc.narg('report_retention_days')::integer, report_retention_days),
+ updated_at = clock_timestamp()
+WHERE id = sqlc.arg('id')
RETURNING *;
-- name: ListWorkspaceMembers :many
-SELECT wm.workspace_id, wm.user_id, wm.role, wm.created_at, wm.updated_at,
- u.username, u.email, u.display_name
-FROM workspace_members AS wm
-JOIN users AS u ON u.id = wm.user_id
-WHERE wm.workspace_id = sqlc.arg(workspace_id)
-ORDER BY wm.created_at, wm.user_id;
-
--- name: UpsertWorkspaceMember :exec
-INSERT INTO workspace_members (workspace_id, user_id, role)
-VALUES (sqlc.arg(workspace_id), sqlc.arg(user_id), sqlc.arg(role))
-ON CONFLICT (workspace_id, user_id) DO UPDATE
-SET role = EXCLUDED.role, updated_at = now();
-
--- name: DeleteWorkspaceMember :execrows
-DELETE FROM workspace_members
-WHERE workspace_id = sqlc.arg(workspace_id) AND user_id = sqlc.arg(user_id);
+SELECT u.id, u.username, u.email, u.display_name, wm.role, wm.created_at
+FROM workspace_members wm JOIN users u ON u.id = wm.user_id
+WHERE wm.workspace_id = sqlc.arg('workspace_id')
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (wm.created_at, wm.user_id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY wm.created_at DESC, wm.user_id DESC
+LIMIT sqlc.arg('limit');
+
+-- name: RemoveWorkspaceMember :execrows
+DELETE FROM workspace_members WHERE workspace_id = $1 AND user_id = $2 AND role <> 'owner';
-- name: CreateWorkspaceInvitation :one
-INSERT INTO workspace_invitations (
- id, workspace_id, email_normalized, role, token_hash, invited_by, expires_at
-) VALUES (
- sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(email_normalized), sqlc.arg(role),
- sqlc.arg(token_hash), sqlc.arg(invited_by), sqlc.arg(expires_at)
-)
+INSERT INTO workspace_invitations (id, workspace_id, email, token_hash, created_by, expires_at)
+VALUES ($1, $2, $3, $4, $5, $6)
RETURNING *;
-- name: ListWorkspaceInvitations :many
SELECT * FROM workspace_invitations
-WHERE workspace_id = sqlc.arg(workspace_id)
-ORDER BY created_at DESC, id DESC;
+WHERE workspace_id = sqlc.arg('workspace_id')
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (created_at, id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY created_at DESC, id DESC
+LIMIT sqlc.arg('limit');
+
+-- name: RevokeWorkspaceInvitation :execrows
+UPDATE workspace_invitations SET revoked_at = now()
+WHERE id = $1 AND workspace_id = $2 AND accepted_at IS NULL AND revoked_at IS NULL;
--- name: GetWorkspaceInvitationForUpdate :one
+-- name: GetInvitationByTokenHash :one
SELECT * FROM workspace_invitations
-WHERE token_hash = sqlc.arg(token_hash)
-FOR UPDATE;
+WHERE token_hash = $1 AND accepted_at IS NULL AND revoked_at IS NULL AND expires_at > now();
-- name: AcceptWorkspaceInvitation :execrows
-UPDATE workspace_invitations SET
- accepted_at = now(),
- accepted_by = sqlc.arg(accepted_by),
- updated_at = now()
-WHERE id = sqlc.arg(id) AND accepted_at IS NULL AND revoked_at IS NULL;
-
--- name: RevokeWorkspaceInvitation :execrows
-UPDATE workspace_invitations SET revoked_at = now(), updated_at = now()
-WHERE id = sqlc.arg(id) AND accepted_at IS NULL AND revoked_at IS NULL;
-
--- name: AppendAccountAudit :exec
-INSERT INTO account_audit_logs (
- id, user_id, actor_user_id, action, result, provider, request_id, metadata, occurred_at
-) VALUES (
- sqlc.arg(id), sqlc.arg(user_id), sqlc.narg(actor_user_id), sqlc.arg(action),
- sqlc.arg(result), sqlc.arg(provider), sqlc.arg(request_id), sqlc.arg(metadata), sqlc.arg(occurred_at)
-);
+UPDATE workspace_invitations SET accepted_at = now()
+WHERE id = $1 AND accepted_at IS NULL AND revoked_at IS NULL;
+-- name: LockWorkspaceQuota :one
+SELECT id FROM workspaces WHERE id=$1 FOR UPDATE;
diff --git a/sql/queries/jobs.sql b/sql/queries/jobs.sql
deleted file mode 100644
index 0debe57..0000000
--- a/sql/queries/jobs.sql
+++ /dev/null
@@ -1,113 +0,0 @@
--- name: CreateJob :one
-INSERT INTO jobs (
- id, workspace_id, type, payload, status, idempotency_key,
- max_attempts, run_after, trace_context
-) VALUES (
- sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(type), sqlc.arg(payload), 'queued',
- sqlc.arg(idempotency_key), sqlc.arg(max_attempts), sqlc.arg(run_after), sqlc.arg(trace_context)
-)
-RETURNING *;
-
--- name: GetJob :one
-SELECT * FROM jobs WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id);
-
--- name: CancelJob :one
-UPDATE jobs
-SET status = 'cancelled', lease_owner = NULL, lease_until = NULL,
- last_error_code = 'job.cancelled', last_error_message = 'Cancelled by user',
- finished_at = now(), updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id)
- AND status IN ('queued', 'running')
-RETURNING *;
-
--- name: CancelActiveRepositorySyncJob :one
-UPDATE jobs
-SET status = 'cancelled', lease_owner = NULL, lease_until = NULL,
- last_error_code = 'job.cancelled', last_error_message = 'Cancelled by user',
- finished_at = now(), updated_at = now()
-WHERE workspace_id = sqlc.arg(workspace_id) AND type = 'repository.sync'
- AND payload->>'repositoryId' = sqlc.arg(repository_id)::text
- AND status IN ('queued', 'running')
-RETURNING *;
-
--- name: ListJobQueueDepths :many
-SELECT type, status, count(*)::bigint AS depth
-FROM jobs
-WHERE type = ANY(sqlc.arg(types)::text[]) AND status IN ('queued', 'running')
-GROUP BY type, status
-ORDER BY type, status;
-
--- name: LockCurrentJob :one
-SELECT id FROM jobs
-WHERE id = sqlc.arg(id) AND status = 'running'
- AND lease_owner = sqlc.arg(lease_owner)
- AND fencing_token = sqlc.arg(fencing_token)
- AND lease_until > now()
-FOR UPDATE;
-
--- name: ClaimJob :one
-WITH candidate AS (
- SELECT id FROM jobs
- WHERE type = ANY(sqlc.arg(types)::text[])
- AND attempt < max_attempts
- AND ((status = 'queued' AND run_after <= now())
- OR (status = 'running' AND lease_until < now()))
- ORDER BY run_after, created_at
- FOR UPDATE SKIP LOCKED
- LIMIT 1
-)
-UPDATE jobs AS job
-SET status = 'running', lease_owner = sqlc.arg(lease_owner),
- lease_until = sqlc.arg(lease_until), fencing_token = fencing_token + 1,
- attempt = attempt + 1, started_at = COALESCE(started_at, now()), updated_at = now()
-FROM candidate
-WHERE job.id = candidate.id
-RETURNING job.*;
-
--- name: RenewJobLease :execrows
-UPDATE jobs SET lease_until = sqlc.arg(lease_until), updated_at = now()
-WHERE id = sqlc.arg(id) AND status = 'running'
- AND lease_owner = sqlc.arg(lease_owner) AND fencing_token = sqlc.arg(fencing_token);
-
--- name: CompleteJob :execrows
-UPDATE jobs SET status = 'succeeded', lease_owner = NULL, lease_until = NULL,
- finished_at = now(), updated_at = now()
-WHERE id = sqlc.arg(id) AND status = 'running'
- AND lease_owner = sqlc.arg(lease_owner) AND fencing_token = sqlc.arg(fencing_token);
-
--- name: RetryJob :execrows
-UPDATE jobs SET status = 'queued', lease_owner = NULL, lease_until = NULL,
- run_after = sqlc.arg(run_after), last_error_code = sqlc.arg(error_code),
- last_error_message = sqlc.arg(error_message), updated_at = now()
-WHERE id = sqlc.arg(id) AND status = 'running' AND attempt < max_attempts
- AND lease_owner = sqlc.arg(lease_owner) AND fencing_token = sqlc.arg(fencing_token);
-
--- name: FailJob :execrows
-UPDATE jobs SET status = 'failed', lease_owner = NULL, lease_until = NULL,
- last_error_code = sqlc.arg(error_code), last_error_message = sqlc.arg(error_message),
- finished_at = now(), updated_at = now()
-WHERE id = sqlc.arg(id) AND status = 'running'
- AND lease_owner = sqlc.arg(lease_owner) AND fencing_token = sqlc.arg(fencing_token);
-
--- name: AppendOutboxEvent :exec
-INSERT INTO outbox_events (
- id, workspace_id, aggregate, aggregate_id, event_type, payload, trace_context
-) VALUES (
- sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(aggregate), sqlc.arg(aggregate_id),
- sqlc.arg(event_type), sqlc.arg(payload), sqlc.arg(trace_context)
-);
-
--- name: AppendAuditLog :exec
-INSERT INTO audit_logs (
- id, workspace_id, actor_user_id, action, resource_type, resource_id, result, metadata
-) VALUES (
- sqlc.arg(id), sqlc.arg(workspace_id), sqlc.narg(actor_user_id), sqlc.arg(action),
- sqlc.arg(resource_type), sqlc.narg(resource_id), sqlc.arg(result), sqlc.arg(metadata)
-);
-
--- name: ListAuditLogs :many
-SELECT * FROM audit_logs
-WHERE workspace_id = sqlc.arg(workspace_id)
- AND (sqlc.narg(before_time)::timestamptz IS NULL OR (occurred_at, id) < (sqlc.narg(before_time), sqlc.narg(before_id)::uuid))
-ORDER BY occurred_at DESC, id DESC
-LIMIT sqlc.arg(page_size);
diff --git a/sql/queries/notifications.sql b/sql/queries/notifications.sql
new file mode 100644
index 0000000..dfb0c03
--- /dev/null
+++ b/sql/queries/notifications.sql
@@ -0,0 +1,44 @@
+-- name: CreateNotification :one
+INSERT INTO notifications (id, workspace_id, analysis_run_id, channel_id, event_type, status)
+VALUES ($1,$2,$3,$4,$5,'queued') RETURNING *;
+
+-- name: StartNotification :one
+UPDATE notifications SET status='running', started_at=now()
+WHERE id=$1 AND status='queued' RETURNING *;
+
+-- name: FinishNotification :execrows
+WITH delivered AS (
+ UPDATE notifications SET status='delivered', finished_at=now()
+ WHERE notifications.id=$1 AND notifications.status='running' RETURNING channel_id
+)
+UPDATE channels SET last_connected_at=now(), last_error_message=NULL, updated_at=now()
+FROM delivered WHERE channels.id=delivered.channel_id;
+
+-- name: FailNotification :exec
+WITH failed AS (
+ UPDATE notifications SET status='failed', error_message=$2, finished_at=now()
+ WHERE notifications.id=$1 AND notifications.status IN ('queued','running') RETURNING channel_id
+)
+UPDATE channels SET last_error_message=$2, updated_at=now()
+FROM failed WHERE channels.id=failed.channel_id;
+
+-- name: CancelNotification :exec
+UPDATE notifications SET status='cancelled', finished_at=now()
+WHERE id=$1 AND status IN ('queued','running');
+
+-- name: CancelChannelNotifications :exec
+WITH cancelled_notifications AS (
+ UPDATE notifications
+ SET status='cancelled', finished_at=now()
+ WHERE channel_id=$1 AND workspace_id=$2 AND status IN ('queued','running')
+ RETURNING id
+)
+UPDATE workflow_dispatches AS dispatch
+SET status='cancelled'
+FROM cancelled_notifications AS notification
+WHERE dispatch.aggregate_type='notification'
+ AND dispatch.aggregate_id=notification.id
+ AND dispatch.status='pending';
+
+-- name: SetNotificationWorkflowID :exec
+UPDATE notifications SET workflow_run_id=$2 WHERE id=$1;
diff --git a/sql/queries/outbox.sql b/sql/queries/outbox.sql
deleted file mode 100644
index 7c4d7f6..0000000
--- a/sql/queries/outbox.sql
+++ /dev/null
@@ -1,33 +0,0 @@
--- name: ClaimOutboxEvent :one
-WITH candidate AS (
- SELECT id
- FROM outbox_events
- WHERE published_at IS NULL AND next_attempt_at <= now()
- AND (lease_until IS NULL OR lease_until < now())
- ORDER BY next_attempt_at, created_at
- FOR UPDATE SKIP LOCKED
- LIMIT 1
-)
-UPDATE outbox_events AS event
-SET lease_owner = sqlc.arg(lease_owner), lease_until = sqlc.arg(lease_until),
- attempt = attempt + 1
-FROM candidate
-WHERE event.id = candidate.id
-RETURNING event.*;
-
--- name: CompleteOutboxEvent :execrows
-UPDATE outbox_events
-SET published_at = now(), lease_owner = NULL, lease_until = NULL,
- last_error_code = '', last_error_message = ''
-WHERE id = sqlc.arg(id) AND published_at IS NULL
- AND lease_owner = sqlc.arg(lease_owner) AND attempt = sqlc.arg(attempt);
-
--- name: RetryOutboxEvent :execrows
-UPDATE outbox_events
-SET lease_owner = NULL, lease_until = NULL, next_attempt_at = sqlc.arg(next_attempt_at),
- last_error_code = sqlc.arg(error_code), last_error_message = sqlc.arg(error_message)
-WHERE id = sqlc.arg(id) AND published_at IS NULL
- AND lease_owner = sqlc.arg(lease_owner) AND attempt = sqlc.arg(attempt);
-
--- name: CountOutboxBacklog :one
-SELECT count(*) FROM outbox_events WHERE published_at IS NULL;
diff --git a/sql/queries/providers.sql b/sql/queries/providers.sql
new file mode 100644
index 0000000..708fc44
--- /dev/null
+++ b/sql/queries/providers.sql
@@ -0,0 +1,54 @@
+-- name: ListProviderConnections :many
+SELECT id, user_id, provider_type, base_url, provider_account_id, login, display_name,
+ scopes, is_default, status, credential_version, last_validated_at, last_used_at, last_error_code,
+ created_at, updated_at
+FROM provider_connections WHERE user_id = $1 AND status <> 'revoked'
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (created_at, id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY created_at DESC, id DESC
+LIMIT sqlc.arg('limit');
+
+-- name: CreateProviderConnection :one
+INSERT INTO provider_connections (
+ id, user_id, provider_type, base_url, token_ciphertext, token_nonce, key_version,
+ provider_account_id, login, display_name, scopes, is_default, status, last_validated_at
+) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,'active',now())
+RETURNING *;
+
+-- name: GetProviderConnection :one
+SELECT * FROM provider_connections WHERE id = $1 AND user_id = $2 AND status <> 'revoked';
+
+-- name: MarkProviderConnectionUsed :execrows
+UPDATE provider_connections SET last_used_at=now(), updated_at=now()
+WHERE id=$1 AND user_id=$2 AND status='active';
+
+-- name: ReplaceProviderCredential :one
+UPDATE provider_connections
+SET token_ciphertext=$3, token_nonce=$4, key_version=$5,
+ credential_version=credential_version+1, status='active', last_error_code=NULL, updated_at=now()
+WHERE id=$1 AND user_id=$2 AND status <> 'revoked'
+RETURNING *;
+
+-- name: SetProviderConnectionValidation :one
+UPDATE provider_connections
+SET provider_account_id=$3, login=$4, display_name=$5, scopes=$6, status=$7,
+ last_validated_at=now(), last_error_code=$8, updated_at=now()
+WHERE id=$1 AND user_id=$2 AND status <> 'revoked'
+RETURNING *;
+
+-- name: ClearDefaultProviderConnections :exec
+UPDATE provider_connections SET is_default=false, updated_at=now()
+WHERE user_id=$1 AND provider_type=$2 AND status <> 'revoked';
+
+-- name: SetDefaultProviderConnection :one
+UPDATE provider_connections SET is_default=true, updated_at=now()
+WHERE id=$1 AND user_id=$2 AND status='active'
+RETURNING *;
+
+-- name: RevokeProviderConnection :execrows
+UPDATE provider_connections
+SET status='revoked', is_default=false, token_ciphertext=decode('', 'hex'),
+ token_nonce=decode('', 'hex'), credential_version=credential_version+1, updated_at=now()
+WHERE id=$1 AND user_id=$2 AND status <> 'revoked';
diff --git a/sql/queries/repositories.sql b/sql/queries/repositories.sql
index 4f0d7c8..7d48c9b 100644
--- a/sql/queries/repositories.sql
+++ b/sql/queries/repositories.sql
@@ -1,102 +1,196 @@
--- name: CreateSCMConnection :one
-INSERT INTO scm_connections (id, workspace_id, type, name, base_url, auth_type, secret_ref)
-VALUES (sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(type), sqlc.arg(name),
- sqlc.arg(base_url), sqlc.arg(auth_type), sqlc.narg(secret_ref))
+-- name: CreateRepository :one
+INSERT INTO repositories (id, workspace_id, provider_type, name, remote_url, normalized_url, configured_ref, git_path, status, created_by)
+VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'provisioning',$9) RETURNING *;
+
+-- name: ReserveRepositorySource :one
+INSERT INTO repository_source_keys (workspace_id, normalized_url, repository_id)
+VALUES ($1,$2,$3)
+ON CONFLICT (workspace_id, normalized_url) DO UPDATE
+SET normalized_url=excluded.normalized_url
+WHERE repository_source_keys.repository_id=excluded.repository_id
+RETURNING repository_id;
+
+-- name: ReleaseRepositorySource :execrows
+DELETE FROM repository_source_keys
+WHERE workspace_id=$1 AND normalized_url=$2 AND repository_id=$3;
+
+-- name: GetRepository :one
+SELECT * FROM repositories WHERE id=$1 AND workspace_id=$2 AND status <> 'deleted';
+
+-- name: ListRepositories :many
+SELECT r.*, s.commit_sha AS current_commit_sha, s.author_name AS current_author_name,
+ s.source_ref AS current_source_ref, s.authored_at AS current_authored_at, s.title AS current_title,
+ s.source_state AS current_source_state, s.created_at AS snapshot_created_at
+FROM repositories r LEFT JOIN commit_snapshots s ON s.id=r.current_snapshot_id
+WHERE r.workspace_id=sqlc.arg('workspace_id')
+ AND r.status <> 'deleted'
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (r.created_at, r.id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY r.created_at DESC, r.id DESC
+LIMIT sqlc.arg('limit');
+
+-- name: UpdateRepository :one
+UPDATE repositories SET name=$3, updated_at=now()
+WHERE id=$1 AND workspace_id=$2 RETURNING *;
+
+-- name: StageRepositoryUpdate :one
+UPDATE repositories
+SET name=sqlc.arg('name'), provider_type=sqlc.arg('provider_type'),
+ remote_url=sqlc.arg('remote_url'), normalized_url=sqlc.arg('normalized_url'),
+ configured_ref=sqlc.arg('configured_ref'), config_version=config_version+1, status='syncing',
+ updated_at=now()
+WHERE id=sqlc.arg('id') AND workspace_id=sqlc.arg('workspace_id')
+ AND config_version=sqlc.arg('config_version')
+ AND status NOT IN ('provisioning','syncing','deleting','deleted')
RETURNING *;
--- name: ListSCMConnections :many
-SELECT * FROM scm_connections
-WHERE workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL
-ORDER BY created_at DESC, id DESC;
+-- name: ArchiveRepository :one
+UPDATE repositories SET archived_at=now(), updated_at=now()
+WHERE id=$1 AND workspace_id=$2 AND archived_at IS NULL RETURNING *;
+
+-- name: RestoreRepository :one
+UPDATE repositories SET archived_at=NULL, updated_at=now()
+WHERE id=$1 AND workspace_id=$2 RETURNING *;
--- name: GetSCMConnection :one
-SELECT * FROM scm_connections
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL;
+-- name: DeleteRepository :execrows
+DELETE FROM repositories WHERE id=$1 AND workspace_id=$2;
--- name: UpdateSCMConnection :one
-UPDATE scm_connections
-SET name = sqlc.arg(name), base_url = sqlc.arg(base_url), auth_type = sqlc.arg(auth_type),
- secret_ref = sqlc.narg(secret_ref), updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL
+-- name: RequestRepositoryDeletion :one
+UPDATE repositories SET status='deleting', updated_at=now()
+WHERE id=$1 AND workspace_id=$2
+ AND status NOT IN ('provisioning','syncing','deleting','deleted')
RETURNING *;
--- name: SoftDeleteSCMConnection :one
-UPDATE scm_connections AS connection
-SET secret_ref = NULL, deleted_at = now(), updated_at = now()
-WHERE connection.id = sqlc.arg(id) AND connection.workspace_id = sqlc.arg(workspace_id)
- AND connection.deleted_at IS NULL
- AND NOT EXISTS (
- SELECT 1 FROM repositories AS repository
- WHERE repository.connection_id = connection.id AND repository.deleted_at IS NULL
+-- name: SetRepositoryStatus :exec
+UPDATE repositories SET status=$2, updated_at=now() WHERE id=$1;
+
+-- name: SetRepositorySyncing :execrows
+UPDATE repositories
+SET status='syncing', updated_at=now()
+WHERE id=$1 AND config_version=$2 AND status NOT IN ('deleting','deleted');
+
+-- name: ProjectRepositoryOperationFailure :execrows
+UPDATE repositories
+SET status=CASE WHEN current_snapshot_id IS NULL THEN 'failed' ELSE 'ready' END,
+ provider_type=sqlc.arg('previous_provider_type'),
+ remote_url=sqlc.arg('previous_remote_url'),
+ normalized_url=sqlc.arg('previous_normalized_url'),
+ configured_ref=sqlc.arg('previous_ref'),
+ last_error_code=sqlc.arg('last_error_code'), last_error_message=sqlc.arg('last_error_message'), updated_at=now()
+WHERE id=sqlc.arg('id') AND config_version=sqlc.arg('config_version') AND status <> 'deleted';
+
+-- name: RestoreDeletedRepositoryAfterPurgeFailure :execrows
+UPDATE repositories
+SET status='deleted', last_error_code=sqlc.arg('last_error_code'),
+ last_error_message=sqlc.arg('last_error_message'), updated_at=now()
+WHERE id=sqlc.arg('id') AND config_version=sqlc.arg('config_version')
+ AND deleted_at IS NOT NULL AND status='deleting';
+
+-- name: RestoreRepositoryAfterOperation :execrows
+UPDATE repositories
+SET status=CASE WHEN current_snapshot_id IS NULL THEN 'failed' ELSE 'ready' END,
+ provider_type=sqlc.arg('previous_provider_type'),
+ remote_url=sqlc.arg('previous_remote_url'),
+ normalized_url=sqlc.arg('previous_normalized_url'),
+ configured_ref=sqlc.arg('previous_ref'),
+ updated_at=now()
+WHERE id=sqlc.arg('id') AND config_version=sqlc.arg('config_version') AND status IN ('syncing','deleting');
+
+-- name: CreateRepositoryOperation :one
+INSERT INTO repository_operations (
+ id, repository_id, actor_user_id, provider_connection_id, credential_version, repository_version, kind,
+ requested_provider_type, requested_remote_url, requested_normalized_url, requested_ref,
+ previous_provider_type, previous_remote_url, previous_normalized_url, previous_ref, status
+) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,'queued') RETURNING *;
+
+-- name: GetRepositoryOperation :one
+SELECT * FROM repository_operations WHERE id=$1;
+
+-- name: ListRepositoryOperations :many
+SELECT * FROM repository_operations
+WHERE repository_id=sqlc.arg('repository_id')
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (created_at, id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
)
-RETURNING connection.*;
+ORDER BY created_at DESC, id DESC
+LIMIT sqlc.arg('limit');
--- name: CreateRepository :one
-INSERT INTO repositories (
- id, workspace_id, connection_id, name, clone_url, normalized_url,
- ref, state
-) VALUES (
- sqlc.arg(id), sqlc.arg(workspace_id), sqlc.narg(connection_id), sqlc.arg(name),
- sqlc.arg(clone_url), sqlc.arg(normalized_url), sqlc.arg(ref), 'pending'
-)
+-- name: StartRepositoryOperation :one
+UPDATE repository_operations SET status='running', started_at=now()
+WHERE id=$1 AND status='queued' RETURNING *;
+
+-- name: FinishRepositoryOperation :one
+UPDATE repository_operations
+SET status='succeeded', outcome=$2, resolved_commit_sha=$3, snapshot_id=$4, finished_at=now()
+WHERE id=$1 AND status='running' RETURNING *;
+
+-- name: FailRepositoryOperation :one
+UPDATE repository_operations SET status='failed', error_message=$2, finished_at=now()
+WHERE id=$1 AND status IN ('queued','running') RETURNING *;
+
+-- name: CancelRepositoryOperation :one
+UPDATE repository_operations SET status='cancelled', finished_at=now()
+WHERE id=$1 AND repository_id=$2 AND status IN ('queued','running')
RETURNING *;
--- name: ListRepositories :many
-SELECT * FROM repositories
-WHERE workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL
-ORDER BY created_at DESC, id DESC;
+-- name: EnsureSnapshot :one
+INSERT INTO commit_snapshots (id, repository_id, commit_sha, source_ref, git_ref, author_name, authored_at, title, source_state)
+VALUES ($1,$2,$3,$4,$5,$6,$7,$8,'available')
+ON CONFLICT (repository_id, commit_sha) DO UPDATE
+SET id=commit_snapshots.id
+RETURNING *;
--- name: GetRepository :one
-SELECT * FROM repositories
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL;
-
--- name: SetRepositoryRef :execrows
-UPDATE repositories SET ref = sqlc.arg(ref), updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL;
-
--- name: CancelRepositorySyncJobs :exec
-UPDATE jobs
-SET status = 'cancelled', lease_owner = NULL, lease_until = NULL,
- last_error_code = 'repository.deleted', last_error_message = 'Repository was deleted',
- finished_at = now(), updated_at = now()
-WHERE workspace_id = sqlc.arg(workspace_id) AND type = 'repository.sync'
- AND payload->>'repositoryId' = sqlc.arg(repository_id)::text
- AND status IN ('queued', 'running');
-
--- name: SoftDeleteRepository :one
+-- name: SetRepositoryCurrentSnapshot :one
UPDATE repositories
-SET state = 'deleting', deleted_at = now(), updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL
-RETURNING *;
+SET current_snapshot_id=sqlc.arg('current_snapshot_id'),
+ provider_type=sqlc.arg('provider_type'), remote_url=sqlc.arg('remote_url'),
+ normalized_url=sqlc.arg('normalized_url'), configured_ref=sqlc.arg('configured_ref'),
+ status='ready', mirror_size_bytes=sqlc.arg('mirror_size_bytes'),
+ last_sync_at=now(), last_error_code=NULL, last_error_message=NULL, updated_at=now()
+WHERE id=sqlc.arg('id') AND workspace_id=sqlc.arg('workspace_id')
+ AND config_version=sqlc.arg('config_version') RETURNING *;
--- name: MarkRepositorySyncing :execrows
-UPDATE repositories SET state = 'syncing', last_error_code = '', last_error_message = '', updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL;
-
--- name: MarkRepositoryFailed :execrows
-UPDATE repositories SET state = 'failed', last_error_code = sqlc.arg(error_code),
- last_error_message = sqlc.arg(error_message), updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL;
-
--- name: MarkRepositoryReady :execrows
-UPDATE repositories SET state = 'ready', current_commit_sha = sqlc.arg(commit_sha), synced_at = now(),
- last_error_code = '', last_error_message = '', updated_at = now()
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id) AND deleted_at IS NULL;
-
--- name: ReserveRepositorySyncCooldown :one
-INSERT INTO repository_sync_controls (repository_id, workspace_id, next_allowed_at)
-VALUES (sqlc.arg(repository_id), sqlc.arg(workspace_id), sqlc.arg(next_allowed_at))
-ON CONFLICT (repository_id) DO UPDATE
-SET next_allowed_at = EXCLUDED.next_allowed_at, updated_at = now()
-WHERE repository_sync_controls.workspace_id = EXCLUDED.workspace_id
- AND repository_sync_controls.next_allowed_at <= now()
-RETURNING repository_id;
+-- name: GetSnapshot :one
+SELECT * FROM commit_snapshots WHERE id=$1 AND repository_id=$2;
+
+-- name: GetWorkspaceSnapshot :one
+SELECT s.* FROM commit_snapshots s
+JOIN repositories r ON r.id=s.repository_id
+WHERE s.id=$1 AND s.repository_id=$2 AND r.workspace_id=$3;
+
+-- name: ListRepositorySnapshots :many
+SELECT s.* FROM commit_snapshots s
+JOIN repositories r ON r.id=s.repository_id
+WHERE s.repository_id=$1 AND r.workspace_id=$2
+ AND (
+ sqlc.narg('cursor_time')::timestamptz IS NULL
+ OR (s.created_at, s.id) < (sqlc.narg('cursor_time'), sqlc.narg('cursor_id')::uuid)
+ )
+ORDER BY s.created_at DESC, s.id DESC
+LIMIT sqlc.arg('limit');
--- name: GetWorkspaceOverview :one
-SELECT
- (SELECT count(*) FROM repositories AS repository
- WHERE repository.workspace_id = sqlc.arg(target_workspace_id) AND repository.deleted_at IS NULL)::bigint AS repository_count,
- (SELECT count(*) FROM channel_instances AS channel
- WHERE channel.workspace_id = sqlc.arg(target_workspace_id) AND channel.enabled = true AND channel.deleted_at IS NULL)::bigint AS active_channel_count,
- (SELECT count(*) FROM jobs AS job
- WHERE job.workspace_id = sqlc.arg(target_workspace_id) AND job.status = 'failed')::bigint AS failed_job_count;
+-- name: MarkSnapshotsPurged :exec
+UPDATE commit_snapshots SET source_state='purged' WHERE repository_id=$1;
+
+-- name: SetRepositoryOperationWorkflowID :exec
+UPDATE repository_operations SET workflow_run_id=$2 WHERE id=$1;
+
+-- name: HasRepositoryAnalysisRuns :one
+SELECT EXISTS (SELECT 1 FROM analysis_runs WHERE repository_id=$1);
+
+-- name: CompleteRepositoryLogicalDeletion :execrows
+WITH completed AS (
+ UPDATE repository_operations
+ SET status='succeeded', outcome='no_change', finished_at=now()
+ WHERE repository_operations.id=$1 AND repository_operations.repository_id=$2 AND repository_operations.status='running'
+ RETURNING repository_operations.repository_id
+)
+UPDATE repositories
+SET status='deleted', archived_at=COALESCE(archived_at, now()), deleted_at=now(), updated_at=now()
+FROM completed
+WHERE repositories.id=completed.repository_id;
+-- name: CountManagedRepositories :one
+SELECT count(*) FROM repositories WHERE workspace_id=$1 AND status <> 'deleted';
diff --git a/sql/queries/retention.sql b/sql/queries/retention.sql
new file mode 100644
index 0000000..dd096ab
--- /dev/null
+++ b/sql/queries/retention.sql
@@ -0,0 +1,154 @@
+-- name: ClaimDueRetentionSchedules :many
+SELECT schedule.workspace_id, schedule.next_run_at, workspace.report_retention_days
+FROM retention_cleanup_schedules AS schedule
+JOIN workspaces AS workspace ON workspace.id = schedule.workspace_id
+WHERE schedule.next_run_at <= clock_timestamp()
+ORDER BY schedule.next_run_at, schedule.workspace_id
+FOR UPDATE OF schedule SKIP LOCKED
+LIMIT sqlc.arg('limit');
+
+-- name: AdvanceRetentionSchedule :execrows
+UPDATE retention_cleanup_schedules
+SET last_run_at = clock_timestamp(),
+ next_run_at = GREATEST(next_run_at + INTERVAL '1 day', clock_timestamp() + INTERVAL '1 day'),
+ updated_at = clock_timestamp()
+WHERE workspace_id = $1 AND next_run_at = $2;
+
+-- name: CreateRetentionCleanup :one
+INSERT INTO retention_cleanups (id, workspace_id, scheduled_for, retention_days, status)
+VALUES ($1, $2, $3, $4, 'queued')
+RETURNING *;
+
+-- name: GetRetentionCleanup :one
+SELECT * FROM retention_cleanups WHERE id = $1;
+
+-- name: StartRetentionCleanup :one
+UPDATE retention_cleanups
+SET status = 'running', started_at = COALESCE(started_at, clock_timestamp()), error_message = NULL
+WHERE id = $1 AND status IN ('queued', 'running')
+RETURNING *;
+
+-- name: DeleteExpiredAnalysisRuns :many
+WITH candidates AS (
+ SELECT analysis_run.id
+ FROM analysis_runs AS analysis_run
+ WHERE analysis_run.workspace_id = sqlc.arg('workspace_id')
+ AND analysis_run.status IN ('succeeded', 'failed', 'cancelled')
+ AND analysis_run.finished_at < clock_timestamp() - make_interval(days => sqlc.arg('retention_days')::integer)
+ ORDER BY analysis_run.finished_at, analysis_run.id
+ FOR UPDATE SKIP LOCKED
+ LIMIT sqlc.arg('limit')
+)
+DELETE FROM analysis_runs AS run
+USING candidates
+WHERE run.id = candidates.id
+RETURNING run.snapshot_id;
+
+-- name: LockPurgeableSnapshotCandidates :many
+SELECT snapshot.id
+FROM commit_snapshots AS snapshot
+JOIN repositories AS repository ON repository.id = snapshot.repository_id
+WHERE repository.workspace_id = sqlc.arg('workspace_id')
+ AND (
+ snapshot.source_state = 'available'
+ OR (
+ snapshot.source_state = 'purging'
+ AND (
+ snapshot.purge_cleanup_id = sqlc.arg('cleanup_id')
+ OR EXISTS (
+ SELECT 1
+ FROM retention_cleanups AS previous_cleanup
+ WHERE previous_cleanup.id = snapshot.purge_cleanup_id
+ AND previous_cleanup.status = 'failed'
+ )
+ )
+ )
+ )
+ AND repository.current_snapshot_id IS DISTINCT FROM snapshot.id
+ AND NOT EXISTS (
+ SELECT 1 FROM analysis_runs AS run WHERE run.snapshot_id = snapshot.id
+ )
+ORDER BY snapshot.created_at, snapshot.id
+FOR UPDATE OF snapshot SKIP LOCKED
+LIMIT sqlc.arg('limit');
+
+-- name: MarkSnapshotsPurging :many
+UPDATE commit_snapshots AS snapshot
+SET source_state = 'purging', purge_cleanup_id = sqlc.arg('cleanup_id')
+FROM repositories AS repository
+WHERE snapshot.id = ANY(sqlc.arg('snapshot_ids')::uuid[])
+ AND repository.id = snapshot.repository_id
+ AND repository.workspace_id = sqlc.arg('workspace_id')
+ AND repository.current_snapshot_id IS DISTINCT FROM snapshot.id
+ AND NOT EXISTS (
+ SELECT 1 FROM analysis_runs AS run WHERE run.snapshot_id = snapshot.id
+ )
+ AND (
+ snapshot.source_state = 'available'
+ OR (
+ snapshot.source_state = 'purging'
+ AND (
+ snapshot.purge_cleanup_id = sqlc.arg('cleanup_id')
+ OR EXISTS (
+ SELECT 1
+ FROM retention_cleanups AS previous_cleanup
+ WHERE previous_cleanup.id = snapshot.purge_cleanup_id
+ AND previous_cleanup.status = 'failed'
+ )
+ )
+ )
+ )
+RETURNING snapshot.id, snapshot.repository_id, snapshot.git_ref, snapshot.commit_sha;
+
+-- name: CompleteSnapshotPurge :execrows
+UPDATE commit_snapshots
+SET source_state = 'purged', purge_cleanup_id = NULL
+WHERE id = $1 AND repository_id = $2 AND purge_cleanup_id = $3 AND source_state = 'purging';
+
+-- name: AddRetentionCleanupProgress :execrows
+UPDATE retention_cleanups
+SET deleted_run_count = deleted_run_count + sqlc.arg('deleted_runs')::integer,
+ purged_snapshot_count = purged_snapshot_count + sqlc.arg('purged_snapshots')::integer
+WHERE id = sqlc.arg('id') AND status = 'running';
+
+-- name: ClaimRepositoriesReadyForPurge :many
+SELECT repository.*
+FROM repositories AS repository
+WHERE repository.workspace_id = sqlc.arg('workspace_id')
+ AND repository.status = 'deleted'
+ AND NOT EXISTS (
+ SELECT 1 FROM analysis_runs AS run WHERE run.repository_id = repository.id
+ )
+ AND NOT EXISTS (
+ SELECT 1 FROM repository_operations AS operation
+ WHERE operation.repository_id = repository.id
+ AND operation.kind = 'purge'
+ AND operation.status IN ('queued', 'running')
+ )
+ORDER BY repository.deleted_at, repository.id
+FOR UPDATE OF repository SKIP LOCKED
+LIMIT sqlc.arg('limit');
+
+-- name: ReactivateRepositoryPurge :execrows
+UPDATE repositories
+SET status = 'deleting', updated_at = clock_timestamp()
+WHERE id = $1 AND status = 'deleted';
+
+-- name: AddRequeuedRepositoryCount :execrows
+UPDATE retention_cleanups
+SET requeued_repository_count = requeued_repository_count + $2
+WHERE id = $1 AND status = 'running';
+
+-- name: FinishRetentionCleanup :one
+UPDATE retention_cleanups
+SET status = 'succeeded', finished_at = clock_timestamp(), error_message = NULL
+WHERE id = $1 AND status = 'running'
+RETURNING *;
+
+-- name: FailRetentionCleanup :execrows
+UPDATE retention_cleanups
+SET status = 'failed', error_message = $2, finished_at = clock_timestamp()
+WHERE id = $1 AND status IN ('queued', 'running');
+
+-- name: SetRetentionCleanupWorkflowID :exec
+UPDATE retention_cleanups SET workflow_run_id = $2 WHERE id = $1;
diff --git a/sql/queries/secrets.sql b/sql/queries/secrets.sql
deleted file mode 100644
index 8ec961e..0000000
--- a/sql/queries/secrets.sql
+++ /dev/null
@@ -1,18 +0,0 @@
--- name: CreateSecret :one
-INSERT INTO secrets (
- id, workspace_id, resource_type, resource_id, ciphertext, nonce,
- wrapped_key, wrapped_key_nonce, key_version
-) VALUES (
- sqlc.arg(id), sqlc.arg(workspace_id), sqlc.arg(resource_type), sqlc.arg(resource_id),
- sqlc.arg(ciphertext), sqlc.arg(nonce), sqlc.arg(wrapped_key),
- sqlc.arg(wrapped_key_nonce), sqlc.arg(key_version)
-)
-RETURNING *;
-
--- name: GetSecret :one
-SELECT * FROM secrets
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id);
-
--- name: DeleteSecret :exec
-DELETE FROM secrets
-WHERE id = sqlc.arg(id) AND workspace_id = sqlc.arg(workspace_id);
diff --git a/sql/queries/workflows.sql b/sql/queries/workflows.sql
new file mode 100644
index 0000000..286d0a4
--- /dev/null
+++ b/sql/queries/workflows.sql
@@ -0,0 +1,34 @@
+-- name: CreateWorkflowDispatch :one
+INSERT INTO workflow_dispatches (id, aggregate_type, aggregate_id, workflow_name, payload, status)
+VALUES ($1,$2,$3,$4,$5,'pending') RETURNING *;
+
+-- name: ClaimWorkflowDispatches :many
+WITH candidates AS (
+ SELECT id FROM workflow_dispatches
+ WHERE status='pending' AND available_at <= now()
+ ORDER BY created_at
+ FOR UPDATE SKIP LOCKED
+ LIMIT sqlc.arg('limit')
+)
+UPDATE workflow_dispatches AS dispatch
+SET attempts=dispatch.attempts+1,
+ available_at=now() + sqlc.arg('lease_duration')::interval
+FROM candidates
+WHERE dispatch.id=candidates.id
+RETURNING dispatch.*;
+
+-- name: MarkWorkflowDispatched :execrows
+UPDATE workflow_dispatches SET status='dispatched', dispatched_at=now()
+WHERE id=$1 AND status='pending';
+
+-- name: DelayWorkflowDispatch :exec
+UPDATE workflow_dispatches
+SET available_at=$2, last_error=$3
+WHERE id=$1 AND status='pending';
+
+-- name: IsWorkflowDispatchPending :one
+SELECT EXISTS(SELECT 1 FROM workflow_dispatches WHERE id=$1 AND status='pending');
+
+-- name: CancelWorkflowDispatch :exec
+UPDATE workflow_dispatches SET status='cancelled'
+WHERE aggregate_type=$1 AND aggregate_id=$2 AND status='pending';
diff --git a/sqlc.yaml b/sqlc.yaml
index fe5550d..df310ec 100644
--- a/sqlc.yaml
+++ b/sqlc.yaml
@@ -1,14 +1,23 @@
version: "2"
sql:
- engine: postgresql
- schema: migrations/postgres
+ schema: migrations/postgres/000001_initial.up.sql
queries: sql/queries
gen:
go:
- package: db
- out: internal/repository/postgres/sqlc
+ package: sqlc
+ out: internal/data/sqlc
sql_package: pgx/v5
emit_interface: true
emit_json_tags: true
emit_empty_slices: true
- emit_pointers_for_null_types: true
+ overrides:
+ - db_type: uuid
+ go_type:
+ import: github.com/google/uuid
+ type: UUID
+ - db_type: uuid
+ nullable: true
+ go_type:
+ import: github.com/google/uuid
+ type: NullUUID
