feat(desktop): drop parked review flow, add publish management

SaaS 侧人工审核后才创建 publish job,desktop 不再做二次审核:移除
manual/waiting_user/parked/from_parked 状态机与 LeaseFromParked 查询,
desktop client 只执行发布并新增"发布管理"页(待发布队列 / 历史 / 再次
发送)。同时抽离 @geo/publisher-platforms 共享适配器包、新增 Redis-based
desktop_presence 与 publish_record_support,刷新 admin-web 发布弹窗与
媒体库;plan A / spec 文档同步口径。
This commit is contained in:
2026-04-20 09:52:48 +08:00
parent b16e9f0bd1
commit a617d39a4a
93 changed files with 5519 additions and 3594 deletions
@@ -3,10 +3,12 @@ package app
import (
"context"
"errors"
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
goredis "github.com/redis/go-redis/v9"
"github.com/geo-platform/tenant-api/internal/shared/auth"
"github.com/geo-platform/tenant-api/internal/shared/response"
@@ -14,11 +16,23 @@ import (
)
type DesktopAccountService struct {
repo repository.DesktopAccountRepository
repo repository.DesktopAccountRepository
clientRepo repository.DesktopClientRepository
redis *goredis.Client
now func() time.Time
}
func NewDesktopAccountService(repo repository.DesktopAccountRepository) *DesktopAccountService {
return &DesktopAccountService{repo: repo}
func NewDesktopAccountService(
repo repository.DesktopAccountRepository,
clientRepo repository.DesktopClientRepository,
redis *goredis.Client,
) *DesktopAccountService {
return &DesktopAccountService{
repo: repo,
clientRepo: clientRepo,
redis: redis,
now: time.Now,
}
}
type DesktopAccountView struct {
@@ -26,6 +40,7 @@ type DesktopAccountView struct {
Platform string `json:"platform"`
PlatformUID string `json:"platform_uid"`
DisplayName string `json:"display_name"`
AvatarURL *string `json:"avatar_url"`
Health string `json:"health"`
ClientID *string `json:"client_id"`
AccountFingerprint *string `json:"account_fingerprint"`
@@ -34,6 +49,9 @@ type DesktopAccountView struct {
SyncVersion int64 `json:"sync_version"`
DeletedAt *time.Time `json:"deleted_at"`
DeleteRequestedAt *time.Time `json:"delete_requested_at"`
ClientOnline *bool `json:"client_online"`
ClientLastSeenAt *time.Time `json:"client_last_seen_at"`
ClientDeviceName *string `json:"client_device_name"`
}
type UpsertDesktopAccountRequest struct {
@@ -41,6 +59,7 @@ type UpsertDesktopAccountRequest struct {
PlatformUID string `json:"platform_uid" binding:"required"`
AccountFingerprint *string `json:"account_fingerprint"`
DisplayName string `json:"display_name" binding:"required"`
AvatarURL *string `json:"avatar_url"`
Health string `json:"health" binding:"required,oneof=live expired captcha risk"`
VerifiedAt *time.Time `json:"verified_at"`
Tags []string `json:"tags"`
@@ -55,19 +74,38 @@ type PatchDesktopAccountRequest struct {
IfSyncVersion int64 `json:"if_sync_version" binding:"required"`
}
func (s *DesktopAccountService) ListByClient(ctx context.Context, client *repository.DesktopClient) ([]DesktopAccountView, error) {
func (s *DesktopAccountService) ListByUser(ctx context.Context, client *repository.DesktopClient) ([]DesktopAccountView, error) {
if client == nil {
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
}
rows, err := s.repo.ListByClient(ctx, client.WorkspaceID, client.ID)
rows, err := s.repo.ListByUser(ctx, client.WorkspaceID, client.UserID)
if err != nil {
return nil, response.ErrInternal(50081, "desktop_accounts_query_failed", "failed to list desktop accounts")
}
clientMap, presenceMap := s.loadClientState(ctx, client.WorkspaceID, rows)
items := make([]DesktopAccountView, 0, len(rows))
for _, item := range rows {
items = append(items, buildDesktopAccountView(&item))
items = append(items, s.buildDesktopAccountView(&item, clientMap, presenceMap))
}
return items, nil
}
func (s *DesktopAccountService) ListForActor(ctx context.Context, actor auth.Actor) ([]DesktopAccountView, error) {
if actor.PrimaryWorkspaceID == 0 || actor.UserID == 0 {
return nil, response.ErrUnauthorized(40132, "workspace_scope_missing", "workspace context is required")
}
rows, err := s.repo.ListByUser(ctx, actor.PrimaryWorkspaceID, actor.UserID)
if err != nil {
return nil, response.ErrInternal(50081, "desktop_accounts_query_failed", "failed to list desktop accounts")
}
clientMap, presenceMap := s.loadClientState(ctx, actor.PrimaryWorkspaceID, rows)
items := make([]DesktopAccountView, 0, len(rows))
for _, item := range rows {
items = append(items, s.buildDesktopAccountView(&item, clientMap, presenceMap))
}
return items, nil
}
@@ -97,6 +135,7 @@ func (s *DesktopAccountService) Upsert(ctx context.Context, client *repository.D
PlatformUID: req.PlatformUID,
AccountFingerprint: req.AccountFingerprint,
DisplayName: req.DisplayName,
AvatarURL: req.AvatarURL,
Health: req.Health,
VerifiedAt: req.VerifiedAt,
Tags: req.Tags,
@@ -109,7 +148,7 @@ func (s *DesktopAccountService) Upsert(ctx context.Context, client *repository.D
return nil, response.ErrInternal(50083, "desktop_account_upsert_failed", "failed to upsert desktop account")
}
view := buildDesktopAccountView(account)
view := s.buildDesktopAccountView(account, nil, nil)
return &view, nil
}
@@ -134,7 +173,7 @@ func (s *DesktopAccountService) Patch(ctx context.Context, client *repository.De
return nil, response.ErrInternal(50084, "desktop_account_patch_failed", "failed to patch desktop account")
}
view := buildDesktopAccountView(account)
view := s.buildDesktopAccountView(account, nil, nil)
return &view, nil
}
@@ -151,7 +190,7 @@ func (s *DesktopAccountService) Tombstone(ctx context.Context, client *repositor
return nil, response.ErrInternal(50085, "desktop_account_delete_failed", "failed to delete desktop account")
}
view := buildDesktopAccountView(account)
view := s.buildDesktopAccountView(account, nil, nil)
return &view, nil
}
@@ -177,22 +216,53 @@ func (s *DesktopAccountService) RequestDelete(ctx context.Context, actor auth.Ac
return nil, response.ErrInternal(50086, "desktop_account_request_delete_failed", "failed to update desktop account delete request")
}
view := buildDesktopAccountView(account)
view := s.buildDesktopAccountView(account, nil, nil)
return &view, nil
}
func buildDesktopAccountView(account *repository.DesktopAccount) DesktopAccountView {
func (s *DesktopAccountService) buildDesktopAccountView(
account *repository.DesktopAccount,
clientMap map[uuid.UUID]*repository.DesktopClient,
presenceClientMap map[uuid.UUID]uuid.UUID,
) DesktopAccountView {
var clientID *string
if account.ClientID != nil {
value := account.ClientID.String()
var clientOnline *bool
var clientLastSeenAt *time.Time
var clientDeviceName *string
resolvedClientID := account.ClientID
if presenceClientMap != nil {
if activeClientID, ok := presenceClientMap[account.DesktopID]; ok {
resolvedClientID = &activeClientID
}
}
if resolvedClientID != nil {
value := resolvedClientID.String()
clientID = &value
online := false
clientOnline = &online
if clientMap != nil {
if client, ok := clientMap[*resolvedClientID]; ok && client != nil {
clientLastSeenAt = client.LastSeenAt
clientDeviceName = client.DeviceName
if presenceClientMap != nil {
if activeClientID, ok := presenceClientMap[account.DesktopID]; ok && activeClientID == *resolvedClientID {
online = true
}
}
}
}
}
return DesktopAccountView{
ID: account.DesktopID.String(),
Platform: account.Platform,
PlatformUID: account.PlatformUID,
PlatformUID: normalizePlatformUID(account.PlatformUID),
DisplayName: account.DisplayName,
AvatarURL: account.AvatarURL,
Health: account.Health,
ClientID: clientID,
AccountFingerprint: account.AccountFingerprint,
@@ -201,5 +271,82 @@ func buildDesktopAccountView(account *repository.DesktopAccount) DesktopAccountV
SyncVersion: account.SyncVersion,
DeletedAt: account.DeletedAt,
DeleteRequestedAt: account.DeleteRequestedAt,
ClientOnline: clientOnline,
ClientLastSeenAt: clientLastSeenAt,
ClientDeviceName: clientDeviceName,
}
}
func (s *DesktopAccountService) loadClientState(
ctx context.Context,
workspaceID int64,
accounts []repository.DesktopAccount,
) (map[uuid.UUID]*repository.DesktopClient, map[uuid.UUID]uuid.UUID) {
clientIDs := make([]uuid.UUID, 0, len(accounts)*2)
accountIDs := make([]uuid.UUID, 0, len(accounts))
seen := make(map[uuid.UUID]struct{}, len(accounts)*2)
for _, account := range accounts {
accountIDs = append(accountIDs, account.DesktopID)
if account.ClientID == nil {
continue
}
if _, ok := seen[*account.ClientID]; ok {
continue
}
seen[*account.ClientID] = struct{}{}
clientIDs = append(clientIDs, *account.ClientID)
}
presenceClientMap := loadDesktopAccountPresence(ctx, s.redis, accountIDs)
for _, clientID := range presenceClientMap {
if _, ok := seen[clientID]; ok {
continue
}
seen[clientID] = struct{}{}
clientIDs = append(clientIDs, clientID)
}
clientMap := make(map[uuid.UUID]*repository.DesktopClient, len(clientIDs))
if s.clientRepo != nil {
for _, clientID := range clientIDs {
client, err := s.clientRepo.GetByID(ctx, clientID, workspaceID)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
continue
}
continue
}
clientMap[clientID] = client
}
}
return clientMap, presenceClientMap
}
func resolveDesktopClientOnline(
client *repository.DesktopClient,
presence bool,
presenceKnown bool,
now time.Time,
) bool {
if client == nil || client.RevokedAt != nil {
return false
}
if presenceKnown && presence {
return true
}
if client.LastSeenAt == nil {
return false
}
return now.UTC().Sub(client.LastSeenAt.UTC()) <= desktopClientPresenceTTL
}
func normalizePlatformUID(value string) string {
normalized := strings.TrimSpace(value)
for strings.HasPrefix(strings.ToLower(normalized), "platform:") {
normalized = strings.TrimSpace(normalized[len("platform:"):])
}
return normalized
}
@@ -0,0 +1,119 @@
package app
import (
"testing"
"time"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/geo-platform/tenant-api/internal/tenant/repository"
)
func TestResolveDesktopClientOnline_PresenceHitWins(t *testing.T) {
now := time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC)
online := resolveDesktopClientOnline(&repository.DesktopClient{
ID: uuid.New(),
LastSeenAt: nil,
}, true, true, now)
assert.True(t, online)
}
func TestResolveDesktopClientOnline_FallsBackToRecentHeartbeatWhenPresenceMisses(t *testing.T) {
now := time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC)
lastSeen := now.Add(-30 * time.Second)
online := resolveDesktopClientOnline(&repository.DesktopClient{
ID: uuid.New(),
LastSeenAt: &lastSeen,
}, false, true, now)
assert.True(t, online)
}
func TestResolveDesktopClientOnline_StaleHeartbeatIsOffline(t *testing.T) {
now := time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC)
lastSeen := now.Add(-(desktopClientPresenceTTL + time.Second))
online := resolveDesktopClientOnline(&repository.DesktopClient{
ID: uuid.New(),
LastSeenAt: &lastSeen,
}, false, false, now)
assert.False(t, online)
}
func TestBuildDesktopAccountView_PrefersAccountPresenceClient(t *testing.T) {
service := &DesktopAccountService{}
accountID := uuid.New()
staleClientID := uuid.New()
activeClientID := uuid.New()
lastSeen := time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC)
deviceName := "Desktop MacIntel"
view := service.buildDesktopAccountView(
&repository.DesktopAccount{
DesktopID: accountID,
ClientID: &staleClientID,
Platform: "toutiaohao",
PlatformUID: "platform:4181862206",
DisplayName: "ireader",
Health: "live",
},
map[uuid.UUID]*repository.DesktopClient{
activeClientID: {
ID: activeClientID,
LastSeenAt: &lastSeen,
DeviceName: &deviceName,
},
},
map[uuid.UUID]uuid.UUID{
accountID: activeClientID,
},
)
if assert.NotNil(t, view.ClientID) {
assert.Equal(t, activeClientID.String(), *view.ClientID)
}
if assert.NotNil(t, view.ClientOnline) {
assert.True(t, *view.ClientOnline)
}
assert.Equal(t, "4181862206", view.PlatformUID)
assert.Equal(t, &lastSeen, view.ClientLastSeenAt)
assert.Equal(t, &deviceName, view.ClientDeviceName)
}
func TestBuildDesktopAccountView_FallsBackToStoredClientWhenNoPresence(t *testing.T) {
service := &DesktopAccountService{}
accountID := uuid.New()
storedClientID := uuid.New()
lastSeen := time.Date(2026, 4, 19, 21, 30, 0, 0, time.UTC)
view := service.buildDesktopAccountView(
&repository.DesktopAccount{
DesktopID: accountID,
ClientID: &storedClientID,
Platform: "toutiaohao",
PlatformUID: "4181862206",
DisplayName: "ireader",
Health: "live",
},
map[uuid.UUID]*repository.DesktopClient{
storedClientID: {
ID: storedClientID,
LastSeenAt: &lastSeen,
},
},
nil,
)
if assert.NotNil(t, view.ClientID) {
assert.Equal(t, storedClientID.String(), *view.ClientID)
}
if assert.NotNil(t, view.ClientOnline) {
assert.False(t, *view.ClientOnline)
}
assert.Equal(t, &lastSeen, view.ClientLastSeenAt)
}
@@ -6,10 +6,12 @@ import (
"crypto/sha256"
"encoding/hex"
"fmt"
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
goredis "github.com/redis/go-redis/v9"
"github.com/geo-platform/tenant-api/internal/shared/auth"
"github.com/geo-platform/tenant-api/internal/shared/response"
@@ -19,14 +21,16 @@ import (
const DesktopClientTokenTTL = 30 * 24 * time.Hour
type DesktopClientService struct {
repo repository.DesktopClientRepository
repo repository.DesktopClientRepository
redis *goredis.Client
}
func NewDesktopClientService(repo repository.DesktopClientRepository) *DesktopClientService {
return &DesktopClientService{repo: repo}
func NewDesktopClientService(repo repository.DesktopClientRepository, redis *goredis.Client) *DesktopClientService {
return &DesktopClientService{repo: repo, redis: redis}
}
type RegisterDesktopClientRequest struct {
ClientID string `json:"client_id"`
DeviceName string `json:"device_name" binding:"required"`
OS string `json:"os" binding:"required"`
CPUArch string `json:"cpu_arch"`
@@ -35,11 +39,12 @@ type RegisterDesktopClientRequest struct {
}
type HeartbeatDesktopClientRequest struct {
DeviceName *string `json:"device_name"`
OS *string `json:"os"`
CPUArch *string `json:"cpu_arch"`
ClientVersion *string `json:"client_version"`
Channel *string `json:"channel"`
DeviceName *string `json:"device_name"`
OS *string `json:"os"`
CPUArch *string `json:"cpu_arch"`
ClientVersion *string `json:"client_version"`
Channel *string `json:"channel"`
AccountIDs []string `json:"account_ids"`
}
type DesktopClientView struct {
@@ -76,6 +81,10 @@ type HeartbeatDesktopClientResponse struct {
ServerTime time.Time `json:"server_time"`
}
type OfflineDesktopClientResponse struct {
ServerTime time.Time `json:"server_time"`
}
func HashDesktopClientToken(raw string) []byte {
sum := sha256.Sum256([]byte(raw))
return sum[:]
@@ -91,8 +100,9 @@ func (s *DesktopClientService) Register(ctx context.Context, actor auth.Actor, r
return nil, response.ErrInternal(50071, "desktop_client_token_failed", "failed to generate desktop client token")
}
clientID := resolveRegisterClientID(req.ClientID)
client, err := s.repo.Register(ctx, repository.RegisterDesktopClientParams{
ID: uuid.New(),
ID: clientID,
TenantID: actor.TenantID,
WorkspaceID: actor.PrimaryWorkspaceID,
UserID: actor.UserID,
@@ -107,6 +117,8 @@ func (s *DesktopClientService) Register(ctx context.Context, actor auth.Actor, r
return nil, response.ErrInternal(50072, "desktop_client_register_failed", "failed to persist desktop client")
}
markDesktopClientPresent(ctx, s.redis, client.ID)
return &RegisterDesktopClientResponse{
ClientID: client.ID.String(),
ClientToken: rawToken,
@@ -137,6 +149,8 @@ func (s *DesktopClientService) Rotate(ctx context.Context, client *repository.De
return nil, response.ErrInternal(50074, "desktop_client_rotate_failed", "failed to rotate desktop client token")
}
markDesktopClientPresent(ctx, s.redis, updated.ID)
return &RotateDesktopClientResponse{
ClientToken: rawToken,
ExpiresAt: time.Now().Add(DesktopClientTokenTTL).Unix(),
@@ -165,6 +179,9 @@ func (s *DesktopClientService) Heartbeat(ctx context.Context, client *repository
return nil, response.ErrInternal(50075, "desktop_client_heartbeat_failed", "failed to persist desktop client heartbeat")
}
markDesktopClientPresent(ctx, s.redis, updated.ID)
markDesktopAccountsPresent(ctx, s.redis, updated.ID, parseDesktopAccountIDs(req.AccountIDs))
return &HeartbeatDesktopClientResponse{
Client: buildDesktopClientView(updated),
ServerTime: time.Now().UTC(),
@@ -184,10 +201,24 @@ func (s *DesktopClientService) Revoke(ctx context.Context, client *repository.De
return nil, response.ErrInternal(50076, "desktop_client_revoke_failed", "failed to revoke desktop client")
}
clearDesktopClientPresenceState(ctx, s.redis, updated.ID)
view := buildDesktopClientView(updated)
return &view, nil
}
func (s *DesktopClientService) Offline(ctx context.Context, client *repository.DesktopClient) (*OfflineDesktopClientResponse, error) {
if client == nil {
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
}
clearDesktopClientPresenceState(ctx, s.redis, client.ID)
return &OfflineDesktopClientResponse{
ServerTime: time.Now().UTC(),
}, nil
}
func newDesktopClientToken() (string, []byte, error) {
raw := make([]byte, 32)
if _, err := rand.Read(raw); err != nil {
@@ -227,3 +258,27 @@ func trimmedStringPtr(value string) *string {
trimmed := value
return &trimmed
}
func parseDesktopAccountIDs(values []string) []uuid.UUID {
if len(values) == 0 {
return nil
}
accountIDs := make([]uuid.UUID, 0, len(values))
for _, value := range values {
parsed, err := uuid.Parse(strings.TrimSpace(value))
if err != nil {
continue
}
accountIDs = append(accountIDs, parsed)
}
return accountIDs
}
func resolveRegisterClientID(value string) uuid.UUID {
parsed, err := uuid.Parse(strings.TrimSpace(value))
if err == nil && parsed != uuid.Nil {
return parsed
}
return uuid.New()
}
@@ -0,0 +1,251 @@
package app
import (
"context"
"strconv"
"time"
"github.com/google/uuid"
goredis "github.com/redis/go-redis/v9"
)
const desktopClientPresenceTTL = 90 * time.Second
const desktopAccountPresenceTTL = 90 * time.Second
func desktopClientPresenceKey(clientID uuid.UUID) string {
return "desktop:presence:client:" + clientID.String()
}
func desktopAccountPresenceKey(accountID uuid.UUID) string {
return "desktop:presence:account:" + accountID.String()
}
func desktopClientAccountsPresenceKey(clientID uuid.UUID) string {
return "desktop:presence:client_accounts:" + clientID.String()
}
func markDesktopClientPresent(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) {
if redis == nil {
return
}
_ = redis.Set(ctx, desktopClientPresenceKey(clientID), time.Now().UTC().Format(time.RFC3339), desktopClientPresenceTTL).Err()
}
func clearDesktopClientPresence(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) {
if redis == nil {
return
}
_ = redis.Del(ctx, desktopClientPresenceKey(clientID)).Err()
}
func loadDesktopClientPresence(ctx context.Context, redis *goredis.Client, clientIDs []uuid.UUID) map[uuid.UUID]bool {
if redis == nil || len(clientIDs) == 0 {
return nil
}
uniqueIDs := make([]uuid.UUID, 0, len(clientIDs))
seen := make(map[uuid.UUID]struct{}, len(clientIDs))
for _, clientID := range clientIDs {
if _, ok := seen[clientID]; ok {
continue
}
seen[clientID] = struct{}{}
uniqueIDs = append(uniqueIDs, clientID)
}
pipe := redis.Pipeline()
cmds := make(map[uuid.UUID]*goredis.IntCmd, len(uniqueIDs))
for _, clientID := range uniqueIDs {
cmds[clientID] = pipe.Exists(ctx, desktopClientPresenceKey(clientID))
}
if _, err := pipe.Exec(ctx); err != nil && err != goredis.Nil {
return nil
}
result := make(map[uuid.UUID]bool, len(cmds))
for clientID, cmd := range cmds {
result[clientID] = cmd.Val() > 0
}
return result
}
func markDesktopAccountsPresent(ctx context.Context, redis *goredis.Client, clientID uuid.UUID, accountIDs []uuid.UUID) {
if redis == nil {
return
}
uniqueIDs := uniqueUUIDs(accountIDs)
trackedAccountIDs := loadTrackedDesktopClientAccountIDs(ctx, redis, clientID)
nextAccountSet := make(map[uuid.UUID]struct{}, len(uniqueIDs))
for _, accountID := range uniqueIDs {
nextAccountSet[accountID] = struct{}{}
}
now := float64(time.Now().UTC().UnixMilli())
staleBefore := strconv.FormatInt(time.Now().UTC().Add(-desktopAccountPresenceTTL).UnixMilli(), 10)
clientAccountsKey := desktopClientAccountsPresenceKey(clientID)
clientMember := clientID.String()
pipe := redis.Pipeline()
pipe.Del(ctx, clientAccountsKey)
if len(uniqueIDs) > 0 {
members := make([]interface{}, 0, len(uniqueIDs))
for _, accountID := range uniqueIDs {
members = append(members, accountID.String())
}
pipe.SAdd(ctx, clientAccountsKey, members...)
pipe.Expire(ctx, clientAccountsKey, desktopAccountPresenceTTL*4)
}
for _, accountID := range trackedAccountIDs {
if _, ok := nextAccountSet[accountID]; ok {
continue
}
pipe.ZRem(ctx, desktopAccountPresenceKey(accountID), clientMember)
}
for _, accountID := range uniqueIDs {
key := desktopAccountPresenceKey(accountID)
pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore)
pipe.ZAdd(ctx, key, goredis.Z{
Score: now,
Member: clientMember,
})
pipe.Expire(ctx, key, desktopAccountPresenceTTL*4)
}
_, _ = pipe.Exec(ctx)
cleanupDesktopAccountPresenceKeys(ctx, redis, trackedAccountIDs)
}
func clearDesktopClientPresenceState(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) {
if redis == nil {
return
}
clearDesktopClientPresence(ctx, redis, clientID)
trackedAccountIDs := loadTrackedDesktopClientAccountIDs(ctx, redis, clientID)
if len(trackedAccountIDs) == 0 {
_ = redis.Del(ctx, desktopClientAccountsPresenceKey(clientID)).Err()
return
}
clientMember := clientID.String()
pipe := redis.Pipeline()
for _, accountID := range trackedAccountIDs {
pipe.ZRem(ctx, desktopAccountPresenceKey(accountID), clientMember)
}
pipe.Del(ctx, desktopClientAccountsPresenceKey(clientID))
_, _ = pipe.Exec(ctx)
cleanupDesktopAccountPresenceKeys(ctx, redis, trackedAccountIDs)
}
func loadDesktopAccountPresence(ctx context.Context, redis *goredis.Client, accountIDs []uuid.UUID) map[uuid.UUID]uuid.UUID {
if redis == nil || len(accountIDs) == 0 {
return nil
}
uniqueIDs := uniqueUUIDs(accountIDs)
staleBefore := strconv.FormatInt(time.Now().UTC().Add(-desktopAccountPresenceTTL).UnixMilli(), 10)
pipe := redis.Pipeline()
commands := make(map[uuid.UUID]*goredis.ZSliceCmd, len(uniqueIDs))
for _, accountID := range uniqueIDs {
key := desktopAccountPresenceKey(accountID)
pipe.ZRemRangeByScore(ctx, key, "-inf", staleBefore)
commands[accountID] = pipe.ZRevRangeWithScores(ctx, key, 0, 0)
}
if _, err := pipe.Exec(ctx); err != nil && err != goredis.Nil {
return nil
}
result := make(map[uuid.UUID]uuid.UUID, len(commands))
for accountID, cmd := range commands {
values := cmd.Val()
if len(values) == 0 {
continue
}
clientIDText, ok := values[0].Member.(string)
if !ok || clientIDText == "" {
continue
}
clientID, err := uuid.Parse(clientIDText)
if err != nil {
continue
}
result[accountID] = clientID
}
return result
}
func uniqueUUIDs(values []uuid.UUID) []uuid.UUID {
if len(values) == 0 {
return nil
}
unique := make([]uuid.UUID, 0, len(values))
seen := make(map[uuid.UUID]struct{}, len(values))
for _, value := range values {
if _, ok := seen[value]; ok {
continue
}
seen[value] = struct{}{}
unique = append(unique, value)
}
return unique
}
func loadTrackedDesktopClientAccountIDs(ctx context.Context, redis *goredis.Client, clientID uuid.UUID) []uuid.UUID {
if redis == nil {
return nil
}
values, err := redis.SMembers(ctx, desktopClientAccountsPresenceKey(clientID)).Result()
if err != nil || len(values) == 0 {
return nil
}
accountIDs := make([]uuid.UUID, 0, len(values))
for _, value := range values {
accountID, parseErr := uuid.Parse(value)
if parseErr != nil {
continue
}
accountIDs = append(accountIDs, accountID)
}
return uniqueUUIDs(accountIDs)
}
func cleanupDesktopAccountPresenceKeys(ctx context.Context, redis *goredis.Client, accountIDs []uuid.UUID) {
if redis == nil || len(accountIDs) == 0 {
return
}
uniqueIDs := uniqueUUIDs(accountIDs)
pipe := redis.Pipeline()
commands := make(map[uuid.UUID]*goredis.IntCmd, len(uniqueIDs))
for _, accountID := range uniqueIDs {
commands[accountID] = pipe.ZCard(ctx, desktopAccountPresenceKey(accountID))
}
if _, err := pipe.Exec(ctx); err != nil && err != goredis.Nil {
return
}
deletePipe := redis.Pipeline()
shouldDelete := false
for accountID, cmd := range commands {
if cmd.Val() > 0 {
continue
}
deletePipe.Del(ctx, desktopAccountPresenceKey(accountID))
shouldDelete = true
}
if shouldDelete {
_, _ = deletePipe.Exec(ctx)
}
}
@@ -9,28 +9,39 @@ import (
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/jackc/pgx/v5/pgxpool"
"go.uber.org/zap"
"github.com/geo-platform/tenant-api/internal/shared/auth"
sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
"github.com/geo-platform/tenant-api/internal/shared/response"
"github.com/geo-platform/tenant-api/internal/tenant/repository"
)
type DesktopTaskService struct {
pool *pgxpool.Pool
repo repository.DesktopTaskRepository
cache sharedcache.Cache
messaging *rabbitmq.Client
logger *zap.Logger
}
func NewDesktopTaskService(repo repository.DesktopTaskRepository, messaging *rabbitmq.Client, logger *zap.Logger) *DesktopTaskService {
func NewDesktopTaskService(pool *pgxpool.Pool, messaging *rabbitmq.Client, logger *zap.Logger) *DesktopTaskService {
return &DesktopTaskService{
repo: repo,
pool: pool,
repo: repository.NewDesktopTaskRepository(pool),
messaging: messaging,
logger: logger,
}
}
func (s *DesktopTaskService) WithCache(c sharedcache.Cache) *DesktopTaskService {
s.cache = c
return s
}
type DesktopTaskView struct {
ID string `json:"id"`
JobID string `json:"job_id"`
@@ -46,7 +57,6 @@ type DesktopTaskView struct {
ActiveAttemptID *string `json:"active_attempt_id"`
LeaseExpiresAt *time.Time `json:"lease_expires_at"`
Attempts int `json:"attempts"`
ParkedReason *string `json:"parked_reason"`
Result json.RawMessage `json:"result"`
Error json.RawMessage `json:"error"`
CreatedAt time.Time `json:"created_at"`
@@ -69,11 +79,6 @@ type ExtendDesktopTaskRequest struct {
LeaseToken string `json:"lease_token" binding:"required"`
}
type ParkDesktopTaskRequest struct {
LeaseToken string `json:"lease_token" binding:"required"`
Reason string `json:"reason" binding:"required,oneof=waiting_user captcha"`
}
type CompleteDesktopTaskRequest struct {
LeaseToken string `json:"lease_token" binding:"required"`
Status string `json:"status" binding:"required,oneof=succeeded failed unknown"`
@@ -92,7 +97,22 @@ type ReconcileDesktopTaskRequest struct {
Error map[string]any `json:"error"`
}
func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.DesktopClient, req LeaseDesktopTaskRequest, routeTaskID *uuid.UUID, fromParked bool) (*LeaseDesktopTaskResponse, error) {
type ListPublishTasksRequest struct {
Page int `form:"page"`
PageSize int `form:"page_size"`
Title string `form:"title"`
}
type DesktopPublishTaskList struct {
Items []DesktopTaskView `json:"items"`
Page int `json:"page"`
PageSize int `json:"page_size"`
Total int `json:"total"`
PendingCount int `json:"pending_count"`
HistoryTotal int `json:"history_total"`
}
func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.DesktopClient, req LeaseDesktopTaskRequest, routeTaskID *uuid.UUID) (*LeaseDesktopTaskResponse, error) {
if client == nil {
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
}
@@ -119,11 +139,6 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt
var task *repository.DesktopTask
switch {
case fromParked:
if taskID == nil {
return nil, response.ErrBadRequest(40085, "missing_desktop_task_id", "task id is required when from_parked=true")
}
task, err = s.repo.LeaseParked(ctx, *taskID, client.ID, leaseParams)
case taskID != nil:
task, err = s.repo.LeaseQueuedByDesktopID(ctx, *taskID, client.ID, leaseParams)
default:
@@ -131,10 +146,10 @@ func (s *DesktopTaskService) Lease(ctx context.Context, client *repository.Deskt
}
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
if taskID == nil && !fromParked {
if taskID == nil {
return &LeaseDesktopTaskResponse{}, nil
}
return nil, s.classifyLeaseError(ctx, client, taskID, fromParked)
return nil, s.classifyLeaseError(ctx, client, taskID)
}
return nil, response.ErrInternal(50088, "desktop_task_lease_failed", "failed to lease desktop task")
}
@@ -179,46 +194,11 @@ func (s *DesktopTaskService) Extend(ctx context.Context, client *repository.Desk
return &view, nil
}
func (s *DesktopTaskService) Park(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, req ParkDesktopTaskRequest) (*DesktopTaskView, error) {
if client == nil {
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
}
previousTask, _ := s.repo.GetByDesktopID(ctx, desktopID, client.WorkspaceID)
previousAttemptID := activeAttemptID(previousTask)
task, err := s.repo.Park(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(req.LeaseToken)), req.Reason)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, response.ErrConflict(40983, "desktop_task_lease_invalid", "desktop task lease token is invalid or expired")
}
return nil, response.ErrInternal(50090, "desktop_task_park_failed", "failed to park desktop task")
}
if previousAttemptID != nil {
if finishErr := s.repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
TaskID: task.DesktopID,
AttemptID: *previousAttemptID,
FinalStatus: literalStringPtr("waiting_user"),
}); finishErr != nil {
s.logWarn("desktop task attempt finish failed after park", finishErr, zap.String("task_id", task.DesktopID.String()))
}
}
s.publishTaskEvent(ctx, task, "task_parked")
view := buildDesktopTaskView(task)
return &view, nil
}
func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.DesktopClient, desktopID uuid.UUID, req CompleteDesktopTaskRequest) (*DesktopTaskView, error) {
if client == nil {
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
}
previousTask, _ := s.repo.GetByDesktopID(ctx, desktopID, client.WorkspaceID)
previousAttemptID := activeAttemptID(previousTask)
resultJSON, err := marshalOptionalJSON(req.Payload)
if err != nil {
return nil, response.ErrBadRequest(40086, "invalid_desktop_task_payload", "payload must be serializable")
@@ -228,7 +208,17 @@ func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.De
return nil, response.ErrBadRequest(40087, "invalid_desktop_task_error", "error must be serializable")
}
task, err := s.repo.Complete(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(req.LeaseToken)), req.Status, resultJSON, errorJSON)
tx, err := s.pool.Begin(ctx)
if err != nil {
return nil, response.ErrInternal(50103, "desktop_task_complete_begin_failed", "failed to start desktop task completion")
}
defer tx.Rollback(ctx)
taskRepo := repository.NewDesktopTaskRepository(tx)
previousTask, _ := taskRepo.GetByDesktopID(ctx, desktopID, client.WorkspaceID)
previousAttemptID := activeAttemptID(previousTask)
task, err := taskRepo.Complete(ctx, desktopID, HashDesktopClientToken(strings.TrimSpace(req.LeaseToken)), req.Status, resultJSON, errorJSON)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, response.ErrConflict(40983, "desktop_task_lease_invalid", "desktop task lease token is invalid or expired")
@@ -237,7 +227,7 @@ func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.De
}
if previousAttemptID != nil {
if finishErr := s.repo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
if finishErr := taskRepo.FinishAttempt(ctx, repository.FinishDesktopTaskAttemptParams{
TaskID: task.DesktopID,
AttemptID: *previousAttemptID,
FinalStatus: &req.Status,
@@ -247,6 +237,19 @@ func (s *DesktopTaskService) Complete(ctx context.Context, client *repository.De
}
}
publishOutcome, err := syncDesktopPublishTaskState(ctx, tx, task)
if err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50104, "desktop_task_complete_commit_failed", "failed to commit desktop task completion")
}
if publishOutcome != nil {
invalidateArticleCaches(ctx, s.cache, publishOutcome.TenantID, &publishOutcome.ArticleID)
}
s.publishTaskEvent(ctx, task, "task_completed")
view := buildDesktopTaskView(task)
@@ -340,7 +343,14 @@ func (s *DesktopTaskService) Reconcile(ctx context.Context, actor auth.Actor, de
return nil, response.ErrBadRequest(40090, "invalid_desktop_task_reconcile_error", "error must be serializable")
}
task, err := s.repo.Reconcile(ctx, desktopID, actor.PrimaryWorkspaceID, req.Status, resultJSON, errorJSON)
tx, err := s.pool.Begin(ctx)
if err != nil {
return nil, response.ErrInternal(50105, "desktop_task_reconcile_begin_failed", "failed to start desktop task reconcile")
}
defer tx.Rollback(ctx)
taskRepo := repository.NewDesktopTaskRepository(tx)
task, err := taskRepo.Reconcile(ctx, desktopID, actor.PrimaryWorkspaceID, req.Status, resultJSON, errorJSON)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, response.ErrConflict(40985, "desktop_task_reconcile_conflict", "desktop task is not waiting for reconcile")
@@ -348,13 +358,280 @@ func (s *DesktopTaskService) Reconcile(ctx context.Context, actor auth.Actor, de
return nil, response.ErrInternal(50094, "desktop_task_reconcile_failed", "failed to reconcile desktop task")
}
publishOutcome, err := syncDesktopPublishTaskState(ctx, tx, task)
if err != nil {
return nil, err
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50106, "desktop_task_reconcile_commit_failed", "failed to commit desktop task reconcile")
}
if publishOutcome != nil {
invalidateArticleCaches(ctx, s.cache, publishOutcome.TenantID, &publishOutcome.ArticleID)
}
s.publishTaskEvent(ctx, task, "task_reconciled")
view := buildDesktopTaskView(task)
return &view, nil
}
func (s *DesktopTaskService) classifyLeaseError(ctx context.Context, client *repository.DesktopClient, taskID *uuid.UUID, fromParked bool) error {
func (s *DesktopTaskService) ListPublishTasks(ctx context.Context, client *repository.DesktopClient, req ListPublishTasksRequest) (*DesktopPublishTaskList, error) {
if client == nil {
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
}
if s.pool == nil {
return nil, response.ErrServiceUnavailable(50342, "desktop_publish_job_store_unavailable", "desktop publish job store is unavailable")
}
page := req.Page
if page <= 0 {
page = 1
}
pageSize := req.PageSize
if pageSize <= 0 || pageSize > 50 {
pageSize = 10
}
title := strings.TrimSpace(req.Title)
pendingStatuses := []string{"queued", "in_progress"}
historyStatuses := []string{"succeeded", "failed", "unknown", "aborted"}
pendingItems, err := s.listPublishTasksByStatuses(
ctx,
client,
pendingStatuses,
title,
0,
0,
`CASE WHEN status = 'in_progress' THEN 0 ELSE 1 END ASC, created_at ASC, desktop_id DESC`,
)
if err != nil {
return nil, err
}
historyTotal, err := s.countPublishTasksByStatuses(ctx, client, historyStatuses, title)
if err != nil {
return nil, err
}
offset := (page - 1) * pageSize
items := make([]DesktopTaskView, 0, pageSize)
total := len(pendingItems) + historyTotal
if offset < len(pendingItems) {
end := offset + pageSize
if end > len(pendingItems) {
end = len(pendingItems)
}
items = append(items, pendingItems[offset:end]...)
}
remaining := pageSize - len(items)
historyOffset := offset - len(pendingItems)
if historyOffset < 0 {
historyOffset = 0
}
if remaining > 0 && historyOffset < historyTotal {
historyItems, listErr := s.listPublishTasksByStatuses(
ctx,
client,
historyStatuses,
title,
remaining,
historyOffset,
`updated_at DESC, desktop_id DESC`,
)
if listErr != nil {
return nil, listErr
}
items = append(items, historyItems...)
}
return &DesktopPublishTaskList{
Items: items,
Page: page,
PageSize: pageSize,
Total: total,
PendingCount: len(pendingItems),
HistoryTotal: historyTotal,
}, nil
}
func (s *DesktopTaskService) listPublishTasksByStatuses(
ctx context.Context,
client *repository.DesktopClient,
statuses []string,
title string,
limit int,
offset int,
orderBy string,
) ([]DesktopTaskView, error) {
query := `
SELECT
desktop_id,
job_id,
tenant_id,
workspace_id,
target_account_id,
target_client_id,
platform_id,
kind,
payload,
status,
dedup_key,
active_attempt_id,
lease_expires_at,
attempts,
result,
error,
created_at,
updated_at
FROM desktop_tasks
WHERE workspace_id = $1
AND target_client_id = $2
AND kind = 'publish'
AND status = ANY($3)
AND ($4 = '' OR COALESCE(payload->>'title', '') ILIKE '%' || $4 || '%')
`
args := []any{client.WorkspaceID, client.ID, statuses, title}
if strings.TrimSpace(orderBy) != "" {
query += "\nORDER BY " + orderBy
}
if limit > 0 {
query += "\nLIMIT $5 OFFSET $6"
args = append(args, limit, offset)
}
rows, err := s.pool.Query(ctx, query, args...)
if err != nil {
return nil, response.ErrInternal(50109, "desktop_publish_tasks_query_failed", "failed to list desktop publish tasks")
}
defer rows.Close()
return scanDesktopTaskRows(rows)
}
func (s *DesktopTaskService) countPublishTasksByStatuses(
ctx context.Context,
client *repository.DesktopClient,
statuses []string,
title string,
) (int, error) {
var count int
err := s.pool.QueryRow(ctx, `
SELECT COUNT(1)
FROM desktop_tasks
WHERE workspace_id = $1
AND target_client_id = $2
AND kind = 'publish'
AND status = ANY($3)
AND ($4 = '' OR COALESCE(payload->>'title', '') ILIKE '%' || $4 || '%')
`, client.WorkspaceID, client.ID, statuses, title).Scan(&count)
if err != nil {
return 0, response.ErrInternal(50109, "desktop_publish_tasks_query_failed", "failed to list desktop publish tasks")
}
return count, nil
}
func scanDesktopTaskRows(rows pgx.Rows) ([]DesktopTaskView, error) {
items := make([]DesktopTaskView, 0)
for rows.Next() {
var (
desktopID uuid.UUID
jobID uuid.UUID
tenantID int64
workspaceID int64
targetAccountID uuid.UUID
targetClientID uuid.UUID
platform string
kind string
payload []byte
status string
dedupKey pgtype.Text
activeAttemptID pgtype.UUID
leaseExpiresAt pgtype.Timestamptz
attempts int32
resultJSON []byte
errorJSON []byte
createdAt time.Time
updatedAt time.Time
)
if err := rows.Scan(
&desktopID,
&jobID,
&tenantID,
&workspaceID,
&targetAccountID,
&targetClientID,
&platform,
&kind,
&payload,
&status,
&dedupKey,
&activeAttemptID,
&leaseExpiresAt,
&attempts,
&resultJSON,
&errorJSON,
&createdAt,
&updatedAt,
); err != nil {
return nil, response.ErrInternal(50110, "desktop_publish_tasks_scan_failed", "failed to scan desktop publish tasks")
}
var dedupKeyText *string
if dedupKey.Valid {
value := dedupKey.String
dedupKeyText = &value
}
var activeAttemptIDText *string
if activeAttemptID.Valid {
value, convErr := uuid.FromBytes(activeAttemptID.Bytes[:])
if convErr == nil {
text := value.String()
activeAttemptIDText = &text
}
}
var leaseExpiresAtValue *time.Time
if leaseExpiresAt.Valid {
value := leaseExpiresAt.Time
leaseExpiresAtValue = &value
}
items = append(items, DesktopTaskView{
ID: desktopID.String(),
JobID: jobID.String(),
TenantID: tenantID,
WorkspaceID: workspaceID,
TargetAccountID: targetAccountID.String(),
TargetClientID: targetClientID.String(),
Platform: platform,
Kind: kind,
Payload: json.RawMessage(payload),
Status: status,
DedupKey: dedupKeyText,
ActiveAttemptID: activeAttemptIDText,
LeaseExpiresAt: leaseExpiresAtValue,
Attempts: int(attempts),
Result: json.RawMessage(resultJSON),
Error: json.RawMessage(errorJSON),
CreatedAt: createdAt,
UpdatedAt: updatedAt,
})
}
if err := rows.Err(); err != nil {
return nil, response.ErrInternal(50109, "desktop_publish_tasks_query_failed", "failed to list desktop publish tasks")
}
return items, nil
}
func (s *DesktopTaskService) classifyLeaseError(ctx context.Context, client *repository.DesktopClient, taskID *uuid.UUID) error {
if taskID == nil {
return response.ErrConflict(40986, "desktop_task_unavailable", "no desktop task is currently available")
}
@@ -371,19 +648,6 @@ func (s *DesktopTaskService) classifyLeaseError(ctx context.Context, client *rep
return response.ErrForbidden(40381, "desktop_task_not_target_client", "desktop task is assigned to another desktop client")
}
if fromParked {
switch task.Status {
case "aborted":
return response.ErrConflict(40987, "desktop_task_aborted", "desktop task has already been aborted")
case "succeeded", "failed", "unknown":
return response.ErrConflict(40988, "desktop_task_terminal", "desktop task is already in a terminal state")
case "in_progress":
return response.ErrConflict(40989, "desktop_task_already_in_progress", "desktop task is already in progress")
default:
return response.ErrConflict(40990, "desktop_task_not_parked", "desktop task is not waiting_user")
}
}
if task.Status == "in_progress" {
return response.ErrConflict(40989, "desktop_task_already_in_progress", "desktop task is already in progress")
}
@@ -413,7 +677,6 @@ func buildDesktopTaskView(task *repository.DesktopTask) DesktopTaskView {
ActiveAttemptID: activeAttemptID,
LeaseExpiresAt: task.LeaseExpiresAt,
Attempts: task.Attempts,
ParkedReason: task.ParkedReason,
Result: json.RawMessage(task.Result),
Error: json.RawMessage(task.Error),
CreatedAt: task.CreatedAt,
+23 -42
View File
@@ -468,51 +468,40 @@ func (s *MediaService) CreatePublishBatch(ctx context.Context, articleID int64,
}
defer tx.Rollback(ctx)
var articleExists bool
var generateStatus string
if err := tx.QueryRow(ctx, `
SELECT EXISTS(SELECT 1 FROM articles WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL),
COALESCE((SELECT generate_status FROM articles WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL), '')
`, articleID, actor.TenantID).Scan(&articleExists, &generateStatus); err != nil {
return nil, response.ErrInternal(50037, "article_check_failed", "failed to validate article")
}
if !articleExists {
return nil, response.ErrNotFound(40411, "article_not_found", "article not found")
}
if generateStatus != "completed" {
return nil, response.ErrConflict(40913, "article_not_publishable", "only completed articles can be published")
articleMeta, err := loadPublishableArticleMeta(ctx, tx, actor.TenantID, articleID)
if err != nil {
return nil, err
}
accountSeeds, err := loadPlatformAccountsForPublish(ctx, tx, actor.TenantID, req.PlatformAccountIDs)
if err != nil {
return nil, err
}
if publishBatchRequiresCover(accountSeeds) && strings.TrimSpace(pointerStringValue(req.CoverAssetURL)) == "" {
coverAssetURL := normalizeStringPointer(req.CoverAssetURL)
if publishBatchRequiresCover(accountSeeds) && strings.TrimSpace(pointerStringValue(coverAssetURL)) == "" {
return nil, response.ErrBadRequest(40044, "publish_cover_required", "baijiahao publishing requires a cover image")
}
var publishBatchID int64
if err := tx.QueryRow(ctx, `
INSERT INTO publish_batches (tenant_id, article_id, initiator_user_id, status, publish_type, cover_asset_url)
VALUES ($1, $2, $3, 'publishing', $4, $5)
RETURNING id
`, actor.TenantID, articleID, actor.UserID, publishType, normalizeStringPointer(req.CoverAssetURL)).Scan(&publishBatchID); err != nil {
return nil, response.ErrInternal(50038, "publish_batch_create_failed", "failed to create publish batch")
if coverAssetURL == nil {
coverAssetURL = articleMeta.CoverAssetURL
}
publishBatchID, publishRecordIDs, err := createPublishBatchRecords(
ctx,
tx,
actor.TenantID,
actor.UserID,
articleID,
publishType,
coverAssetURL,
accountSeeds,
)
if err != nil {
return nil, err
}
tasks := make([]PublishBatchTask, 0, len(accountSeeds))
for _, account := range accountSeeds {
var publishRecordID int64
if err := tx.QueryRow(ctx, `
INSERT INTO publish_records (
tenant_id, publish_batch_id, article_id, platform_account_id, platform_id, status
)
VALUES ($1, $2, $3, $4, $5, 'queued')
RETURNING id
`, actor.TenantID, publishBatchID, articleID, account.ID, account.PlatformID).Scan(&publishRecordID); err != nil {
return nil, response.ErrInternal(50039, "publish_record_create_failed", "failed to create publish record")
}
token, tokenErr := newSessionToken()
if tokenErr != nil {
return nil, response.ErrInternal(50034, "session_token_failed", "failed to generate plugin session token")
@@ -527,12 +516,12 @@ func (s *MediaService) CreatePublishBatch(ctx context.Context, articleID int64,
)
VALUES ($1, $2, $3, 'publish', 'publish_record', $4, $5, $6, $7, $8, 'pending')
RETURNING id
`, actor.TenantID, actor.UserID, req.PluginInstallationID, publishRecordID, account.ID, account.PlatformID, token, expireAt).Scan(&pluginSessionID); err != nil {
`, actor.TenantID, actor.UserID, req.PluginInstallationID, publishRecordIDs[account.ID], account.ID, account.PlatformID, token, expireAt).Scan(&pluginSessionID); err != nil {
return nil, response.ErrInternal(50040, "publish_session_create_failed", "failed to create publish session")
}
tasks = append(tasks, PublishBatchTask{
PublishRecordID: publishRecordID,
PublishRecordID: publishRecordIDs[account.ID],
PlatformAccountID: account.ID,
PlatformID: account.PlatformID,
PlatformName: account.PlatformName,
@@ -544,14 +533,6 @@ func (s *MediaService) CreatePublishBatch(ctx context.Context, articleID int64,
})
}
if _, err := tx.Exec(ctx, `
UPDATE articles
SET publish_status = 'publishing', updated_at = NOW()
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
`, articleID, actor.TenantID); err != nil {
return nil, response.ErrInternal(50041, "article_publish_status_failed", "failed to update article publish status")
}
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50042, "publish_batch_commit_failed", "failed to commit publish batch")
}
+181 -27
View File
@@ -9,9 +9,11 @@ import (
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
goredis "github.com/redis/go-redis/v9"
"go.uber.org/zap"
"github.com/geo-platform/tenant-api/internal/shared/auth"
sharedcache "github.com/geo-platform/tenant-api/internal/shared/cache"
"github.com/geo-platform/tenant-api/internal/shared/messaging/rabbitmq"
"github.com/geo-platform/tenant-api/internal/shared/response"
"github.com/geo-platform/tenant-api/internal/tenant/repository"
@@ -20,20 +22,32 @@ import (
type PublishJobService struct {
pool *pgxpool.Pool
messaging *rabbitmq.Client
redis *goredis.Client
logger *zap.Logger
cache sharedcache.Cache
}
func NewPublishJobService(pool *pgxpool.Pool, messaging *rabbitmq.Client, logger *zap.Logger) *PublishJobService {
func NewPublishJobService(
pool *pgxpool.Pool,
messaging *rabbitmq.Client,
redis *goredis.Client,
logger *zap.Logger,
) *PublishJobService {
return &PublishJobService{
pool: pool,
messaging: messaging,
redis: redis,
logger: logger,
}
}
func (s *PublishJobService) WithCache(c sharedcache.Cache) *PublishJobService {
s.cache = c
return s
}
type CreatePublishJobAccountRequest struct {
AccountID string `json:"account_id" binding:"required"`
Mode string `json:"mode" binding:"required,oneof=auto manual"`
}
type CreatePublishJobRequest struct {
@@ -61,6 +75,10 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
if err != nil {
return nil, response.ErrBadRequest(40091, "invalid_content_ref", "content_ref must be serializable")
}
articleID, err := resolvePublishArticleID(req.ContentRef)
if err != nil {
return nil, err
}
tx, err := s.pool.Begin(ctx)
if err != nil {
@@ -70,6 +88,50 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
accountRepo := repository.NewDesktopAccountRepository(tx)
taskRepo := repository.NewDesktopTaskRepository(tx)
articleMeta, err := loadPublishableArticleMeta(ctx, tx, actor.TenantID, articleID)
if err != nil {
return nil, err
}
type publishTarget struct {
account *repository.DesktopAccount
accountSeed *platformAccountSeed
targetClientID uuid.UUID
}
targets := make([]publishTarget, 0, len(req.Accounts))
for _, accountReq := range req.Accounts {
accountID, parseErr := uuid.Parse(strings.TrimSpace(accountReq.AccountID))
if parseErr != nil {
return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "account_id must be a uuid")
}
account, lookupErr := accountRepo.GetByDesktopID(ctx, accountID, actor.PrimaryWorkspaceID)
if lookupErr != nil {
if errors.Is(lookupErr, pgx.ErrNoRows) {
return nil, response.ErrBadRequest(40093, "desktop_account_not_found", "one or more selected desktop accounts do not exist")
}
return nil, response.ErrInternal(50098, "desktop_account_lookup_failed", "failed to load selected desktop account")
}
if account.Health != "live" {
return nil, response.ErrConflict(40993, "desktop_account_not_publishable", "one or more selected desktop accounts require re-authorization before publishing")
}
targetClientID := s.resolveTargetClientID(ctx, account)
if targetClientID == nil {
return nil, response.ErrConflict(40992, "desktop_account_client_missing", "one or more selected desktop accounts are not currently available on any desktop client")
}
accountSeed, seedErr := loadPlatformAccountSeedByDesktopID(ctx, tx, actor.TenantID, actor.PrimaryWorkspaceID, account.DesktopID)
if seedErr != nil {
return nil, seedErr
}
targets = append(targets, publishTarget{
account: account,
accountSeed: accountSeed,
targetClientID: *targetClientID,
})
}
jobID := uuid.New()
job, err := taskRepo.CreatePublishJob(ctx, repository.CreateDesktopPublishJobParams{
@@ -85,31 +147,41 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
return nil, response.ErrInternal(50097, "desktop_publish_job_create_failed", "failed to create desktop publish job")
}
taskIDs := make([]string, 0, len(req.Accounts))
createdTasks := make([]*repository.DesktopTask, 0, len(req.Accounts))
for _, accountReq := range req.Accounts {
accountID, parseErr := uuid.Parse(strings.TrimSpace(accountReq.AccountID))
if parseErr != nil {
return nil, response.ErrBadRequest(40092, "invalid_desktop_account_id", "account_id must be a uuid")
}
accountSeeds := make([]platformAccountSeed, 0, len(targets))
for _, target := range targets {
accountSeeds = append(accountSeeds, *target.accountSeed)
}
if publishBatchRequiresCover(accountSeeds) && strings.TrimSpace(pointerStringValue(articleMeta.CoverAssetURL)) == "" {
return nil, response.ErrBadRequest(40044, "publish_cover_required", "baijiahao publishing requires a cover image")
}
account, lookupErr := accountRepo.GetByDesktopID(ctx, accountID, actor.PrimaryWorkspaceID)
if lookupErr != nil {
if errors.Is(lookupErr, pgx.ErrNoRows) {
return nil, response.ErrBadRequest(40093, "desktop_account_not_found", "one or more selected desktop accounts do not exist")
}
return nil, response.ErrInternal(50098, "desktop_account_lookup_failed", "failed to load selected desktop account")
}
if account.ClientID == nil {
return nil, response.ErrConflict(40992, "desktop_account_client_missing", "one or more selected desktop accounts are not currently bound to a desktop client")
}
publishBatchID, publishRecordIDs, err := createPublishBatchRecords(
ctx,
tx,
actor.TenantID,
actor.UserID,
articleID,
"publish",
articleMeta.CoverAssetURL,
accountSeeds,
)
if err != nil {
return nil, err
}
taskIDs := make([]string, 0, len(targets))
createdTasks := make([]*repository.DesktopTask, 0, len(targets))
for _, target := range targets {
publishRecordID := publishRecordIDs[target.accountSeed.ID]
payloadJSON, payloadErr := marshalOptionalJSON(map[string]any{
"title": req.Title,
"content_ref": req.ContentRef,
"mode": accountReq.Mode,
"account_id": account.DesktopID.String(),
"platform": account.Platform,
"title": req.Title,
"content_ref": req.ContentRef,
"account_id": target.account.DesktopID.String(),
"platform": target.account.Platform,
"article_id": articleID,
"publish_batch_id": publishBatchID,
"publish_record_id": publishRecordID,
"platform_account_id": target.accountSeed.ID,
})
if payloadErr != nil {
return nil, response.ErrBadRequest(40094, "invalid_desktop_task_payload", "publish job payload must be serializable")
@@ -120,9 +192,9 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
JobID: job.DesktopID,
TenantID: actor.TenantID,
WorkspaceID: actor.PrimaryWorkspaceID,
TargetAccountID: account.DesktopID,
TargetClientID: *account.ClientID,
Platform: account.Platform,
TargetAccountID: target.account.DesktopID,
TargetClientID: target.targetClientID,
Platform: target.account.Platform,
Kind: "publish",
Payload: payloadJSON,
Status: "queued",
@@ -138,6 +210,7 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
if err := tx.Commit(ctx); err != nil {
return nil, response.ErrInternal(50100, "desktop_publish_job_commit_failed", "failed to commit desktop publish job")
}
invalidateArticleCaches(ctx, s.cache, actor.TenantID, &articleID)
for _, task := range createdTasks {
if publishErr := publishDesktopTaskEvent(context.Background(), s.messaging, DesktopTaskEvent{
@@ -160,6 +233,62 @@ func (s *PublishJobService) Create(ctx context.Context, actor auth.Actor, req Cr
}, nil
}
func (s *PublishJobService) RetryByDesktopTask(
ctx context.Context,
client *repository.DesktopClient,
taskID uuid.UUID,
) (*CreatePublishJobResponse, error) {
if client == nil {
return nil, response.ErrUnauthorized(40108, "desktop_client_missing", "desktop client context is required")
}
if s.pool == nil {
return nil, response.ErrServiceUnavailable(50342, "desktop_publish_job_store_unavailable", "desktop publish job store is unavailable")
}
taskRepo := repository.NewDesktopTaskRepository(s.pool)
task, err := taskRepo.GetByDesktopID(ctx, taskID, client.WorkspaceID)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, response.ErrNotFound(40482, "desktop_task_not_found", "desktop task not found")
}
return nil, response.ErrInternal(50095, "desktop_task_lookup_failed", "failed to inspect desktop task state")
}
if task.Kind != "publish" {
return nil, response.ErrBadRequest(40095, "desktop_task_not_publish", "only publish tasks can be retried")
}
if task.TargetClientID != client.ID {
return nil, response.ErrForbidden(40381, "desktop_task_not_target_client", "desktop task is assigned to another desktop client")
}
if task.Status == "queued" || task.Status == "in_progress" {
return nil, response.ErrConflict(40994, "desktop_task_retry_conflict", "desktop task is still pending or running")
}
payload := unmarshalJSONObject(task.Payload)
articleID, ok := extractInt64FromMap(payload, "article_id")
if !ok || articleID <= 0 {
return nil, response.ErrBadRequest(40096, "desktop_task_article_missing", "desktop publish task is missing article_id")
}
title := strings.TrimSpace(stringPointerValue(extractStringPointer(payload, "title")))
if title == "" {
title = "文章发布"
}
return s.Create(ctx, auth.Actor{
TenantID: client.TenantID,
UserID: client.UserID,
PrimaryWorkspaceID: client.WorkspaceID,
}, CreatePublishJobRequest{
Title: title,
ContentRef: map[string]any{
"article_id": articleID,
},
Accounts: []CreatePublishJobAccountRequest{
{AccountID: task.TargetAccountID.String()},
},
})
}
func (s *PublishJobService) logWarn(message string, err error, fields ...zap.Field) {
if s.logger == nil || err == nil {
return
@@ -167,3 +296,28 @@ func (s *PublishJobService) logWarn(message string, err error, fields ...zap.Fie
fields = append(fields, zap.Error(err))
s.logger.Warn(message, fields...)
}
func (s *PublishJobService) resolveTargetClientID(ctx context.Context, account *repository.DesktopAccount) *uuid.UUID {
if account == nil {
return nil
}
presence := loadDesktopAccountPresence(ctx, s.redis, []uuid.UUID{account.DesktopID})
if clientID, ok := presence[account.DesktopID]; ok {
return &clientID
}
if account.ClientID == nil {
return nil
}
clientID := *account.ClientID
return &clientID
}
func stringPointerValue(value *string) string {
if value == nil {
return ""
}
return strings.TrimSpace(*value)
}
@@ -0,0 +1,358 @@
package app
import (
"context"
"encoding/json"
"errors"
"strconv"
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/geo-platform/tenant-api/internal/shared/response"
"github.com/geo-platform/tenant-api/internal/tenant/repository"
)
type publishableArticleMeta struct {
CoverAssetURL *string
}
type desktopPublishSyncOutcome struct {
TenantID int64
ArticleID int64
}
func loadPublishableArticleMeta(ctx context.Context, tx pgx.Tx, tenantID, articleID int64) (*publishableArticleMeta, error) {
var generateStatus string
var wizardStateJSON []byte
if err := tx.QueryRow(ctx, `
SELECT generate_status, wizard_state_json
FROM articles
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
`, articleID, tenantID).Scan(&generateStatus, &wizardStateJSON); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, response.ErrNotFound(40411, "article_not_found", "article not found")
}
return nil, response.ErrInternal(50037, "article_check_failed", "failed to validate article")
}
if generateStatus != "completed" {
return nil, response.ErrConflict(40913, "article_not_publishable", "only completed articles can be published")
}
return &publishableArticleMeta{
CoverAssetURL: resolveArticleCoverAssetURL(wizardStateJSON),
}, nil
}
func createPublishBatchRecords(
ctx context.Context,
tx pgx.Tx,
tenantID int64,
initiatorUserID int64,
articleID int64,
publishType string,
coverAssetURL *string,
accounts []platformAccountSeed,
) (int64, map[int64]int64, error) {
var publishBatchID int64
if err := tx.QueryRow(ctx, `
INSERT INTO publish_batches (tenant_id, article_id, initiator_user_id, status, publish_type, cover_asset_url)
VALUES ($1, $2, $3, 'publishing', $4, $5)
RETURNING id
`, tenantID, articleID, initiatorUserID, publishType, normalizeStringPointer(coverAssetURL)).Scan(&publishBatchID); err != nil {
return 0, nil, response.ErrInternal(50038, "publish_batch_create_failed", "failed to create publish batch")
}
recordIDs := make(map[int64]int64, len(accounts))
for _, account := range accounts {
var publishRecordID int64
if err := tx.QueryRow(ctx, `
INSERT INTO publish_records (
tenant_id, publish_batch_id, article_id, platform_account_id, platform_id, status
)
VALUES ($1, $2, $3, $4, $5, 'queued')
RETURNING id
`, tenantID, publishBatchID, articleID, account.ID, account.PlatformID).Scan(&publishRecordID); err != nil {
return 0, nil, response.ErrInternal(50039, "publish_record_create_failed", "failed to create publish record")
}
recordIDs[account.ID] = publishRecordID
}
if _, err := tx.Exec(ctx, `
UPDATE articles
SET publish_status = 'publishing', updated_at = NOW()
WHERE id = $1 AND tenant_id = $2 AND deleted_at IS NULL
`, articleID, tenantID); err != nil {
return 0, nil, response.ErrInternal(50041, "article_publish_status_failed", "failed to update article publish status")
}
return publishBatchID, recordIDs, nil
}
func syncDesktopPublishTaskState(
ctx context.Context,
tx pgx.Tx,
task *repository.DesktopTask,
) (*desktopPublishSyncOutcome, error) {
if task == nil || task.Kind != "publish" {
return nil, nil
}
payload := unmarshalJSONObject(task.Payload)
publishRecordID, ok := extractInt64FromMap(payload, "publish_record_id")
if !ok || publishRecordID <= 0 {
return nil, nil
}
var publishBatchID int64
var articleID int64
if err := tx.QueryRow(ctx, `
SELECT publish_batch_id, article_id
FROM publish_records
WHERE id = $1 AND tenant_id = $2
`, publishRecordID, task.TenantID).Scan(&publishBatchID, &articleID); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, response.ErrNotFound(40434, "publish_record_not_found", "publish record not found")
}
return nil, response.ErrInternal(50101, "desktop_publish_record_lookup_failed", "failed to lookup desktop publish record")
}
resultPayload := unmarshalJSONObject(task.Result)
errorPayload := unmarshalJSONObject(task.Error)
publishStatus := desktopTaskToPublishStatus(task.Status)
var publishedAt *time.Time
if publishStatus == "success" {
now := time.Now().UTC()
publishedAt = &now
}
if _, err := tx.Exec(ctx, `
UPDATE publish_records
SET status = $1,
external_article_id = $2,
external_article_url = $3,
external_manage_url = $4,
published_at = $5,
request_payload_json = $6,
response_payload_json = $7,
error_message = $8,
updated_at = NOW()
WHERE id = $9 AND tenant_id = $10
`, publishStatus,
extractStringPointer(resultPayload, "external_article_id", "externalArticleId"),
extractStringPointer(resultPayload, "external_article_url", "externalArticleUrl"),
extractStringPointer(resultPayload, "external_manage_url", "externalManageUrl", "review_url", "reviewUrl"),
publishedAt,
nullableJSON(task.Payload),
nullableJSON(task.Result),
desktopTaskErrorMessage(errorPayload),
publishRecordID,
task.TenantID,
); err != nil {
return nil, response.ErrInternal(50102, "desktop_publish_record_update_failed", "failed to update desktop publish record")
}
batchStatus, err := recalculatePublishBatchStatus(ctx, tx, publishBatchID)
if err != nil {
return nil, err
}
if _, err := tx.Exec(ctx, `
UPDATE publish_batches
SET status = $1, updated_at = NOW()
WHERE id = $2
`, batchStatus, publishBatchID); err != nil {
return nil, response.ErrInternal(50054, "publish_batch_update_failed", "failed to update publish batch")
}
articleStatus := publishBatchStatusToArticleStatus(batchStatus)
if _, err := tx.Exec(ctx, `
UPDATE articles
SET publish_status = $1, updated_at = NOW()
WHERE id = $2 AND tenant_id = $3 AND deleted_at IS NULL
`, articleStatus, articleID, task.TenantID); err != nil {
return nil, response.ErrInternal(50055, "article_publish_update_failed", "failed to update article publish status")
}
return &desktopPublishSyncOutcome{
TenantID: task.TenantID,
ArticleID: articleID,
}, nil
}
func loadPlatformAccountSeedByDesktopID(
ctx context.Context,
tx pgx.Tx,
tenantID int64,
workspaceID int64,
desktopID uuid.UUID,
) (*platformAccountSeed, error) {
var item platformAccountSeed
if err := tx.QueryRow(ctx, `
SELECT pa.id, pa.platform_id, mp.name, pa.platform_uid, pa.nickname
FROM platform_accounts pa
JOIN media_platforms mp ON mp.platform_id = pa.platform_id
WHERE pa.tenant_id = $1
AND pa.workspace_id = $2
AND pa.desktop_id = $3
AND pa.deleted_at IS NULL
LIMIT 1
`, tenantID, workspaceID, desktopID).Scan(
&item.ID,
&item.PlatformID,
&item.PlatformName,
&item.PlatformUID,
&item.Nickname,
); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, response.ErrBadRequest(40037, "invalid_platform_accounts", "one or more selected platform accounts are unavailable")
}
return nil, response.ErrInternal(50058, "platform_account_publish_query_failed", "failed to load platform accounts")
}
return &item, nil
}
func desktopTaskToPublishStatus(taskStatus string) string {
switch strings.TrimSpace(taskStatus) {
case "queued":
return "queued"
case "in_progress":
return "publishing"
case "unknown":
return "pending_review"
case "succeeded":
return "success"
case "failed", "aborted":
return "failed"
default:
return "failed"
}
}
func publishBatchStatusToArticleStatus(batchStatus string) string {
normalized := normalizePublishStatus(batchStatus)
switch normalized {
case "success":
return "success"
case "failed":
return "failed"
default:
return normalized
}
}
func resolvePublishArticleID(contentRef map[string]any) (int64, error) {
if articleID, ok := extractInt64FromMap(contentRef, "article_id", "id"); ok && articleID > 0 {
return articleID, nil
}
return 0, response.ErrBadRequest(40091, "invalid_content_ref", "content_ref.article_id is required")
}
func unmarshalJSONObject(raw []byte) map[string]any {
if len(raw) == 0 {
return nil
}
var payload map[string]any
if err := json.Unmarshal(raw, &payload); err != nil {
return nil
}
return payload
}
func nullableJSON(raw []byte) []byte {
if len(raw) == 0 {
return nil
}
return raw
}
func extractInt64FromMap(payload map[string]any, keys ...string) (int64, bool) {
if len(payload) == 0 {
return 0, false
}
for _, key := range keys {
value, ok := extractInt64FromAny(payload[key])
if ok {
return value, true
}
}
return 0, false
}
func extractInt64FromAny(value any) (int64, bool) {
switch typed := value.(type) {
case nil:
return 0, false
case int:
return int64(typed), true
case int8:
return int64(typed), true
case int16:
return int64(typed), true
case int32:
return int64(typed), true
case int64:
return typed, true
case float32:
return int64(typed), true
case float64:
return int64(typed), true
case json.Number:
parsed, err := typed.Int64()
if err != nil {
return 0, false
}
return parsed, true
case string:
trimmed := strings.TrimSpace(typed)
if trimmed == "" {
return 0, false
}
parsed, err := strconv.ParseInt(trimmed, 10, 64)
if err != nil {
return 0, false
}
return parsed, true
default:
return 0, false
}
}
func extractStringPointer(payload map[string]any, keys ...string) *string {
if len(payload) == 0 {
return nil
}
for _, key := range keys {
value, ok := payload[key]
if !ok {
continue
}
switch typed := value.(type) {
case string:
trimmed := strings.TrimSpace(typed)
if trimmed != "" {
return &trimmed
}
case json.Number:
trimmed := strings.TrimSpace(typed.String())
if trimmed != "" {
return &trimmed
}
}
}
return nil
}
func desktopTaskErrorMessage(payload map[string]any) *string {
if message := extractStringPointer(payload, "message"); message != nil {
return message
}
if detail := extractStringPointer(payload, "detail"); detail != nil {
return detail
}
if code := extractStringPointer(payload, "code"); code != nil {
return code
}
return nil
}
@@ -21,6 +21,7 @@ type DesktopAccount struct {
PlatformUID string
AccountFingerprint *string
DisplayName string
AvatarURL *string
Health string
VerifiedAt *time.Time
Tags []string
@@ -41,6 +42,7 @@ type UpsertDesktopAccountParams struct {
PlatformUID string
AccountFingerprint *string
DisplayName string
AvatarURL *string
Health string
VerifiedAt *time.Time
Tags []string
@@ -58,7 +60,7 @@ type PatchDesktopAccountParams struct {
}
type DesktopAccountRepository interface {
ListByClient(ctx context.Context, workspaceID int64, clientID uuid.UUID) ([]DesktopAccount, error)
ListByUser(ctx context.Context, workspaceID, userID int64) ([]DesktopAccount, error)
GetByDesktopID(ctx context.Context, desktopID uuid.UUID, workspaceID int64) (*DesktopAccount, error)
GetByIdentity(ctx context.Context, workspaceID int64, platform, platformUID string) (*DesktopAccount, error)
Upsert(ctx context.Context, params UpsertDesktopAccountParams) (*DesktopAccount, error)
@@ -76,10 +78,10 @@ func NewDesktopAccountRepository(db generated.DBTX) DesktopAccountRepository {
return &desktopAccountRepository{q: newQuerier(db)}
}
func (r *desktopAccountRepository) ListByClient(ctx context.Context, workspaceID int64, clientID uuid.UUID) ([]DesktopAccount, error) {
rows, err := r.q.ListDesktopAccountsByClient(ctx, generated.ListDesktopAccountsByClientParams{
func (r *desktopAccountRepository) ListByUser(ctx context.Context, workspaceID, userID int64) ([]DesktopAccount, error) {
rows, err := r.q.ListDesktopAccountsByUser(ctx, generated.ListDesktopAccountsByUserParams{
WorkspaceID: workspaceID,
ClientID: pgUUID(clientID),
UserID: userID,
})
if err != nil {
return nil, err
@@ -142,6 +144,7 @@ func (r *desktopAccountRepository) Upsert(ctx context.Context, params UpsertDesk
PlatformID: params.Platform,
PlatformUid: params.PlatformUID,
DisplayName: params.DisplayName,
AvatarUrl: pgText(params.AvatarURL),
LegacyStatus: legacyStatusFromHealth(params.Health),
MetadataJson: metadataJSON,
VerifiedAt: pgTimestamp(params.VerifiedAt),
@@ -216,7 +219,7 @@ func (r *desktopAccountRepository) ClearDeleteRequested(ctx context.Context, des
return desktopAccountFromClearDeleteRow(row)
}
func desktopAccountFromListRow(row generated.ListDesktopAccountsByClientRow) (*DesktopAccount, error) {
func desktopAccountFromListRow(row generated.ListDesktopAccountsByUserRow) (*DesktopAccount, error) {
return buildDesktopAccount(
row.DesktopID,
row.TenantID,
@@ -227,6 +230,7 @@ func desktopAccountFromListRow(row generated.ListDesktopAccountsByClientRow) (*D
row.PlatformUid,
row.AccountFingerprint,
row.DisplayName,
row.AvatarUrl,
row.Health,
row.VerifiedAt,
row.Tags,
@@ -249,6 +253,7 @@ func desktopAccountFromGetRow(row generated.GetDesktopAccountByDesktopIDRow) (*D
row.PlatformUid,
row.AccountFingerprint,
row.DisplayName,
row.AvatarUrl,
row.Health,
row.VerifiedAt,
row.Tags,
@@ -271,6 +276,7 @@ func desktopAccountFromIdentityRow(row generated.GetDesktopAccountByIdentityRow)
row.PlatformUid,
row.AccountFingerprint,
row.DisplayName,
row.AvatarUrl,
row.Health,
row.VerifiedAt,
row.Tags,
@@ -293,6 +299,7 @@ func desktopAccountFromUpsertRow(row generated.UpsertDesktopAccountRow) (*Deskto
row.PlatformUid,
row.AccountFingerprint,
row.DisplayName,
row.AvatarUrl,
row.Health,
row.VerifiedAt,
row.Tags,
@@ -315,6 +322,7 @@ func desktopAccountFromPatchRow(row generated.PatchDesktopAccountRow) (*DesktopA
row.PlatformUid,
row.AccountFingerprint,
row.DisplayName,
row.AvatarUrl,
row.Health,
row.VerifiedAt,
row.Tags,
@@ -337,6 +345,7 @@ func desktopAccountFromTombstoneRow(row generated.TombstoneDesktopAccountRow) (*
row.PlatformUid,
row.AccountFingerprint,
row.DisplayName,
row.AvatarUrl,
row.Health,
row.VerifiedAt,
row.Tags,
@@ -359,6 +368,7 @@ func desktopAccountFromMarkDeleteRow(row generated.MarkDesktopAccountDeleteReque
row.PlatformUid,
row.AccountFingerprint,
row.DisplayName,
row.AvatarUrl,
row.Health,
row.VerifiedAt,
row.Tags,
@@ -381,6 +391,7 @@ func desktopAccountFromClearDeleteRow(row generated.ClearDesktopAccountDeleteReq
row.PlatformUid,
row.AccountFingerprint,
row.DisplayName,
row.AvatarUrl,
row.Health,
row.VerifiedAt,
row.Tags,
@@ -402,6 +413,7 @@ func buildDesktopAccount(
platformUID string,
accountFingerprint pgtype.Text,
displayName string,
avatarURL pgtype.Text,
health string,
verifiedAt pgtype.Timestamptz,
tags []byte,
@@ -426,6 +438,7 @@ func buildDesktopAccount(
PlatformUID: platformUID,
AccountFingerprint: nullableText(accountFingerprint),
DisplayName: displayName,
AvatarURL: nullableText(avatarURL),
Health: health,
VerifiedAt: optionalTime(verifiedAt),
Tags: tagList,
@@ -57,6 +57,7 @@ type RotateDesktopClientTokenParams struct {
type DesktopClientRepository interface {
Register(ctx context.Context, params RegisterDesktopClientParams) (*DesktopClient, error)
GetByID(ctx context.Context, id uuid.UUID, workspaceID int64) (*DesktopClient, error)
GetByTokenHash(ctx context.Context, tokenHash []byte) (*DesktopClient, error)
RotateToken(ctx context.Context, params RotateDesktopClientTokenParams) (*DesktopClient, error)
Heartbeat(ctx context.Context, params HeartbeatDesktopClientParams) (*DesktopClient, error)
@@ -91,6 +92,17 @@ func (r *desktopClientRepository) Register(ctx context.Context, params RegisterD
return desktopClientFromGenerated(row), nil
}
func (r *desktopClientRepository) GetByID(ctx context.Context, id uuid.UUID, workspaceID int64) (*DesktopClient, error) {
row, err := r.q.GetDesktopClientByID(ctx, generated.GetDesktopClientByIDParams{
ID: pgUUID(id),
WorkspaceID: workspaceID,
})
if err != nil {
return nil, err
}
return desktopClientFromGenerated(row), nil
}
func (r *desktopClientRepository) GetByTokenHash(ctx context.Context, tokenHash []byte) (*DesktopClient, error) {
row, err := r.q.GetDesktopClientByTokenHash(ctx, tokenHash)
if err != nil {
@@ -46,7 +46,6 @@ type DesktopTask struct {
ActiveAttemptID *uuid.UUID
LeaseExpiresAt *time.Time
Attempts int
ParkedReason *string
Result []byte
Error []byte
CreatedAt time.Time
@@ -65,7 +64,6 @@ type CreateDesktopTaskParams struct {
Payload []byte
Status string
DedupKey *string
ParkedReason *string
}
type DesktopTaskLeaseParams struct {
@@ -104,9 +102,7 @@ type DesktopTaskRepository interface {
GetByDesktopID(ctx context.Context, desktopID uuid.UUID, workspaceID int64) (*DesktopTask, error)
LeaseNextQueued(ctx context.Context, clientID uuid.UUID, kind *string, params DesktopTaskLeaseParams) (*DesktopTask, error)
LeaseQueuedByDesktopID(ctx context.Context, desktopID, clientID uuid.UUID, params DesktopTaskLeaseParams) (*DesktopTask, error)
LeaseParked(ctx context.Context, desktopID, clientID uuid.UUID, params DesktopTaskLeaseParams) (*DesktopTask, error)
ExtendLease(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte) (*DesktopTask, error)
Park(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte, parkedReason string) (*DesktopTask, error)
Complete(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte, status string, resultJSON, errorJSON []byte) (*DesktopTask, error)
CancelByLease(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte, errorJSON []byte) (*DesktopTask, error)
CancelByClient(ctx context.Context, desktopID, clientID uuid.UUID, errorJSON []byte) (*DesktopTask, error)
@@ -154,7 +150,6 @@ func (r *desktopTaskRepository) CreateTask(ctx context.Context, params CreateDes
Payload: params.Payload,
Status: params.Status,
DedupKey: pgText(params.DedupKey),
ParkedReason: pgText(params.ParkedReason),
})
if err != nil {
return nil, err
@@ -199,19 +194,6 @@ func (r *desktopTaskRepository) LeaseQueuedByDesktopID(ctx context.Context, desk
return desktopTaskFromGenerated(row), nil
}
func (r *desktopTaskRepository) LeaseParked(ctx context.Context, desktopID, clientID uuid.UUID, params DesktopTaskLeaseParams) (*DesktopTask, error) {
row, err := r.q.LeaseParkedDesktopTask(ctx, generated.LeaseParkedDesktopTaskParams{
AttemptID: pgUUID(params.AttemptID),
LeaseTokenHash: params.LeaseTokenHash,
DesktopID: pgUUID(desktopID),
ClientID: pgUUID(clientID),
})
if err != nil {
return nil, err
}
return desktopTaskFromGenerated(row), nil
}
func (r *desktopTaskRepository) ExtendLease(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte) (*DesktopTask, error) {
row, err := r.q.ExtendDesktopTaskLease(ctx, generated.ExtendDesktopTaskLeaseParams{
DesktopID: pgUUID(desktopID),
@@ -223,18 +205,6 @@ func (r *desktopTaskRepository) ExtendLease(ctx context.Context, desktopID uuid.
return desktopTaskFromGenerated(row), nil
}
func (r *desktopTaskRepository) Park(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte, parkedReason string) (*DesktopTask, error) {
row, err := r.q.ParkDesktopTask(ctx, generated.ParkDesktopTaskParams{
ParkedReason: pgText(&parkedReason),
DesktopID: pgUUID(desktopID),
LeaseTokenHash: leaseTokenHash,
})
if err != nil {
return nil, err
}
return desktopTaskFromGenerated(row), nil
}
func (r *desktopTaskRepository) Complete(ctx context.Context, desktopID uuid.UUID, leaseTokenHash []byte, status string, resultJSON, errorJSON []byte) (*DesktopTask, error) {
row, err := r.q.CompleteDesktopTask(ctx, generated.CompleteDesktopTaskParams{
Status: status,
@@ -359,7 +329,6 @@ func desktopTaskFromGenerated(row generated.DesktopTask) *DesktopTask {
ActiveAttemptID: nullableUUID(row.ActiveAttemptID),
LeaseExpiresAt: optionalTime(row.LeaseExpiresAt),
Attempts: int(row.Attempts),
ParkedReason: nullableText(row.ParkedReason),
Result: row.Result,
Error: row.Error,
CreatedAt: timeFromTimestamp(row.CreatedAt),
@@ -29,6 +29,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -54,6 +55,7 @@ type ClearDesktopAccountDeleteRequestedRow struct {
PlatformUid string `json:"platform_uid"`
AccountFingerprint pgtype.Text `json:"account_fingerprint"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
Health string `json:"health"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
Tags []byte `json:"tags"`
@@ -77,6 +79,7 @@ func (q *Queries) ClearDesktopAccountDeleteRequested(ctx context.Context, arg Cl
&i.PlatformUid,
&i.AccountFingerprint,
&i.DisplayName,
&i.AvatarUrl,
&i.Health,
&i.VerifiedAt,
&i.Tags,
@@ -100,6 +103,7 @@ SELECT
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -129,6 +133,7 @@ type GetDesktopAccountByDesktopIDRow struct {
PlatformUid string `json:"platform_uid"`
AccountFingerprint pgtype.Text `json:"account_fingerprint"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
Health string `json:"health"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
Tags []byte `json:"tags"`
@@ -152,6 +157,7 @@ func (q *Queries) GetDesktopAccountByDesktopID(ctx context.Context, arg GetDeskt
&i.PlatformUid,
&i.AccountFingerprint,
&i.DisplayName,
&i.AvatarUrl,
&i.Health,
&i.VerifiedAt,
&i.Tags,
@@ -175,6 +181,7 @@ SELECT
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -207,6 +214,7 @@ type GetDesktopAccountByIdentityRow struct {
PlatformUid string `json:"platform_uid"`
AccountFingerprint pgtype.Text `json:"account_fingerprint"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
Health string `json:"health"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
Tags []byte `json:"tags"`
@@ -230,6 +238,7 @@ func (q *Queries) GetDesktopAccountByIdentity(ctx context.Context, arg GetDeskto
&i.PlatformUid,
&i.AccountFingerprint,
&i.DisplayName,
&i.AvatarUrl,
&i.Health,
&i.VerifiedAt,
&i.Tags,
@@ -242,7 +251,7 @@ func (q *Queries) GetDesktopAccountByIdentity(ctx context.Context, arg GetDeskto
return i, err
}
const listDesktopAccountsByClient = `-- name: ListDesktopAccountsByClient :many
const listDesktopAccountsByUser = `-- name: ListDesktopAccountsByUser :many
SELECT
desktop_id,
tenant_id,
@@ -253,6 +262,7 @@ SELECT
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -263,16 +273,17 @@ SELECT
updated_at
FROM platform_accounts
WHERE workspace_id = $1
AND client_id = $2
AND user_id = $2
AND deleted_at IS NULL
ORDER BY platform_id, display_name, created_at DESC
`
type ListDesktopAccountsByClientParams struct {
WorkspaceID int64 `json:"workspace_id"`
ClientID pgtype.UUID `json:"client_id"`
type ListDesktopAccountsByUserParams struct {
WorkspaceID int64 `json:"workspace_id"`
UserID int64 `json:"user_id"`
}
type ListDesktopAccountsByClientRow struct {
type ListDesktopAccountsByUserRow struct {
DesktopID pgtype.UUID `json:"desktop_id"`
TenantID int64 `json:"tenant_id"`
WorkspaceID int64 `json:"workspace_id"`
@@ -282,6 +293,7 @@ type ListDesktopAccountsByClientRow struct {
PlatformUid string `json:"platform_uid"`
AccountFingerprint pgtype.Text `json:"account_fingerprint"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
Health string `json:"health"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
Tags []byte `json:"tags"`
@@ -292,15 +304,15 @@ type ListDesktopAccountsByClientRow struct {
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
}
func (q *Queries) ListDesktopAccountsByClient(ctx context.Context, arg ListDesktopAccountsByClientParams) ([]ListDesktopAccountsByClientRow, error) {
rows, err := q.db.Query(ctx, listDesktopAccountsByClient, arg.WorkspaceID, arg.ClientID)
func (q *Queries) ListDesktopAccountsByUser(ctx context.Context, arg ListDesktopAccountsByUserParams) ([]ListDesktopAccountsByUserRow, error) {
rows, err := q.db.Query(ctx, listDesktopAccountsByUser, arg.WorkspaceID, arg.UserID)
if err != nil {
return nil, err
}
defer rows.Close()
var items []ListDesktopAccountsByClientRow
var items []ListDesktopAccountsByUserRow
for rows.Next() {
var i ListDesktopAccountsByClientRow
var i ListDesktopAccountsByUserRow
if err := rows.Scan(
&i.DesktopID,
&i.TenantID,
@@ -311,6 +323,7 @@ func (q *Queries) ListDesktopAccountsByClient(ctx context.Context, arg ListDeskt
&i.PlatformUid,
&i.AccountFingerprint,
&i.DisplayName,
&i.AvatarUrl,
&i.Health,
&i.VerifiedAt,
&i.Tags,
@@ -348,6 +361,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -373,6 +387,7 @@ type MarkDesktopAccountDeleteRequestedRow struct {
PlatformUid string `json:"platform_uid"`
AccountFingerprint pgtype.Text `json:"account_fingerprint"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
Health string `json:"health"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
Tags []byte `json:"tags"`
@@ -396,6 +411,7 @@ func (q *Queries) MarkDesktopAccountDeleteRequested(ctx context.Context, arg Mar
&i.PlatformUid,
&i.AccountFingerprint,
&i.DisplayName,
&i.AvatarUrl,
&i.Health,
&i.VerifiedAt,
&i.Tags,
@@ -433,6 +449,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -464,6 +481,7 @@ type PatchDesktopAccountRow struct {
PlatformUid string `json:"platform_uid"`
AccountFingerprint pgtype.Text `json:"account_fingerprint"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
Health string `json:"health"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
Tags []byte `json:"tags"`
@@ -496,6 +514,7 @@ func (q *Queries) PatchDesktopAccount(ctx context.Context, arg PatchDesktopAccou
&i.PlatformUid,
&i.AccountFingerprint,
&i.DisplayName,
&i.AvatarUrl,
&i.Health,
&i.VerifiedAt,
&i.Tags,
@@ -527,6 +546,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -553,6 +573,7 @@ type TombstoneDesktopAccountRow struct {
PlatformUid string `json:"platform_uid"`
AccountFingerprint pgtype.Text `json:"account_fingerprint"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
Health string `json:"health"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
Tags []byte `json:"tags"`
@@ -576,6 +597,7 @@ func (q *Queries) TombstoneDesktopAccount(ctx context.Context, arg TombstoneDesk
&i.PlatformUid,
&i.AccountFingerprint,
&i.DisplayName,
&i.AvatarUrl,
&i.Health,
&i.VerifiedAt,
&i.Tags,
@@ -598,6 +620,7 @@ INSERT INTO platform_accounts (
platform_id,
platform_uid,
nickname,
avatar_url,
status,
metadata_json,
last_check_at,
@@ -621,10 +644,11 @@ VALUES (
$10,
$11,
$12,
$8::text,
$13,
$11,
$8::text,
$14,
$12,
$15,
1
)
ON CONFLICT (workspace_id, platform_id, platform_uid) WHERE deleted_at IS NULL
@@ -632,6 +656,7 @@ DO UPDATE SET
client_id = EXCLUDED.client_id,
user_id = EXCLUDED.user_id,
nickname = EXCLUDED.nickname,
avatar_url = COALESCE(EXCLUDED.avatar_url, platform_accounts.avatar_url),
status = EXCLUDED.status,
metadata_json = COALESCE(EXCLUDED.metadata_json, platform_accounts.metadata_json),
last_check_at = COALESCE(EXCLUDED.last_check_at, platform_accounts.last_check_at),
@@ -645,8 +670,8 @@ DO UPDATE SET
updated_at = now()
WHERE platform_accounts.workspace_id = EXCLUDED.workspace_id
AND (
$15::bigint IS NULL
OR platform_accounts.sync_version = $15::bigint
$16::bigint IS NULL
OR platform_accounts.sync_version = $16::bigint
)
RETURNING
desktop_id,
@@ -658,6 +683,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -677,6 +703,7 @@ type UpsertDesktopAccountParams struct {
PlatformID string `json:"platform_id"`
PlatformUid string `json:"platform_uid"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
LegacyStatus string `json:"legacy_status"`
MetadataJson []byte `json:"metadata_json"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
@@ -696,6 +723,7 @@ type UpsertDesktopAccountRow struct {
PlatformUid string `json:"platform_uid"`
AccountFingerprint pgtype.Text `json:"account_fingerprint"`
DisplayName string `json:"display_name"`
AvatarUrl pgtype.Text `json:"avatar_url"`
Health string `json:"health"`
VerifiedAt pgtype.Timestamptz `json:"verified_at"`
Tags []byte `json:"tags"`
@@ -716,6 +744,7 @@ func (q *Queries) UpsertDesktopAccount(ctx context.Context, arg UpsertDesktopAcc
arg.PlatformID,
arg.PlatformUid,
arg.DisplayName,
arg.AvatarUrl,
arg.LegacyStatus,
arg.MetadataJson,
arg.VerifiedAt,
@@ -735,6 +764,7 @@ func (q *Queries) UpsertDesktopAccount(ctx context.Context, arg UpsertDesktopAcc
&i.PlatformUid,
&i.AccountFingerprint,
&i.DisplayName,
&i.AvatarUrl,
&i.Health,
&i.VerifiedAt,
&i.Tags,
@@ -11,6 +11,42 @@ import (
"github.com/jackc/pgx/v5/pgtype"
)
const getDesktopClientByID = `-- name: GetDesktopClientByID :one
SELECT id, tenant_id, workspace_id, user_id, token_hash, device_name, os, cpu_arch, client_version, channel, created_at, last_seen_at, last_rotated_at, revoked_at
FROM desktop_clients
WHERE id = $1
AND workspace_id = $2
AND revoked_at IS NULL
LIMIT 1
`
type GetDesktopClientByIDParams struct {
ID pgtype.UUID `json:"id"`
WorkspaceID int64 `json:"workspace_id"`
}
func (q *Queries) GetDesktopClientByID(ctx context.Context, arg GetDesktopClientByIDParams) (DesktopClient, error) {
row := q.db.QueryRow(ctx, getDesktopClientByID, arg.ID, arg.WorkspaceID)
var i DesktopClient
err := row.Scan(
&i.ID,
&i.TenantID,
&i.WorkspaceID,
&i.UserID,
&i.TokenHash,
&i.DeviceName,
&i.Os,
&i.CpuArch,
&i.ClientVersion,
&i.Channel,
&i.CreatedAt,
&i.LastSeenAt,
&i.LastRotatedAt,
&i.RevokedAt,
)
return i, err
}
const getDesktopClientByTokenHash = `-- name: GetDesktopClientByTokenHash :one
SELECT id, tenant_id, workspace_id, user_id, token_hash, device_name, os, cpu_arch, client_version, channel, created_at, last_seen_at, last_rotated_at, revoked_at
FROM desktop_clients
@@ -163,6 +199,19 @@ VALUES (
$9,
$10
)
ON CONFLICT (id) DO UPDATE
SET tenant_id = EXCLUDED.tenant_id,
workspace_id = EXCLUDED.workspace_id,
user_id = EXCLUDED.user_id,
token_hash = EXCLUDED.token_hash,
device_name = EXCLUDED.device_name,
os = EXCLUDED.os,
cpu_arch = EXCLUDED.cpu_arch,
client_version = EXCLUDED.client_version,
channel = EXCLUDED.channel,
last_seen_at = now(),
last_rotated_at = now(),
revoked_at = NULL
RETURNING id, tenant_id, workspace_id, user_id, token_hash, device_name, os, cpu_arch, client_version, channel, created_at, last_seen_at, last_rotated_at, revoked_at
`
@@ -18,12 +18,11 @@ SET status = 'aborted',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = $2
AND target_client_id = $3
AND status IN ('queued', 'waiting_user')
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
AND status = 'queued'
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
`
type CancelDesktopTaskByClientParams struct {
@@ -52,7 +51,6 @@ func (q *Queries) CancelDesktopTaskByClient(ctx context.Context, arg CancelDeskt
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -68,12 +66,11 @@ SET status = 'aborted',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = $2
AND lease_token_hash = $3
AND status = 'in_progress'
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
`
type CancelDesktopTaskByLeaseParams struct {
@@ -102,7 +99,6 @@ func (q *Queries) CancelDesktopTaskByLease(ctx context.Context, arg CancelDeskto
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -119,12 +115,11 @@ SET status = $1,
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = $4
AND lease_token_hash = $5
AND status = 'in_progress'
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
`
type CompleteDesktopTaskParams struct {
@@ -161,7 +156,6 @@ func (q *Queries) CompleteDesktopTask(ctx context.Context, arg CompleteDesktopTa
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -240,8 +234,7 @@ INSERT INTO desktop_tasks (
kind,
payload,
status,
dedup_key,
parked_reason
dedup_key
)
VALUES (
$1,
@@ -254,10 +247,9 @@ VALUES (
$8,
$9,
$10,
$11,
$12
$11
)
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
`
type CreateDesktopTaskParams struct {
@@ -272,7 +264,6 @@ type CreateDesktopTaskParams struct {
Payload []byte `json:"payload"`
Status string `json:"status"`
DedupKey pgtype.Text `json:"dedup_key"`
ParkedReason pgtype.Text `json:"parked_reason"`
}
func (q *Queries) CreateDesktopTask(ctx context.Context, arg CreateDesktopTaskParams) (DesktopTask, error) {
@@ -288,7 +279,6 @@ func (q *Queries) CreateDesktopTask(ctx context.Context, arg CreateDesktopTaskPa
arg.Payload,
arg.Status,
arg.DedupKey,
arg.ParkedReason,
)
var i DesktopTask
err := row.Scan(
@@ -308,7 +298,6 @@ func (q *Queries) CreateDesktopTask(ctx context.Context, arg CreateDesktopTaskPa
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -370,7 +359,7 @@ SET lease_expires_at = now() + interval '10 minutes',
WHERE desktop_id = $1
AND lease_token_hash = $2
AND status = 'in_progress'
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
`
type ExtendDesktopTaskLeaseParams struct {
@@ -398,7 +387,6 @@ func (q *Queries) ExtendDesktopTaskLease(ctx context.Context, arg ExtendDesktopT
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -434,7 +422,7 @@ func (q *Queries) FinishDesktopTaskAttempt(ctx context.Context, arg FinishDeskto
}
const getDesktopTaskByDesktopID = `-- name: GetDesktopTaskByDesktopID :one
SELECT id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
SELECT id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
FROM desktop_tasks
WHERE desktop_id = $1
AND workspace_id = $2
@@ -466,7 +454,6 @@ func (q *Queries) GetDesktopTaskByDesktopID(ctx context.Context, arg GetDesktopT
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -495,11 +482,10 @@ SET active_attempt_id = $1,
lease_expires_at = now() + interval '10 minutes',
status = 'in_progress',
attempts = t.attempts + 1,
parked_reason = NULL,
updated_at = now()
FROM candidate
WHERE t.desktop_id = candidate.desktop_id
RETURNING t.id, t.desktop_id, t.job_id, t.tenant_id, t.workspace_id, t.target_account_id, t.target_client_id, t.platform_id, t.kind, t.payload, t.status, t.dedup_key, t.active_attempt_id, t.lease_token_hash, t.lease_expires_at, t.attempts, t.parked_reason, t.result, t.error, t.created_at, t.updated_at
RETURNING t.id, t.desktop_id, t.job_id, t.tenant_id, t.workspace_id, t.target_account_id, t.target_client_id, t.platform_id, t.kind, t.payload, t.status, t.dedup_key, t.active_attempt_id, t.lease_token_hash, t.lease_expires_at, t.attempts, t.result, t.error, t.created_at, t.updated_at
`
type LeaseNextQueuedDesktopTaskParams struct {
@@ -534,64 +520,6 @@ func (q *Queries) LeaseNextQueuedDesktopTask(ctx context.Context, arg LeaseNextQ
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const leaseParkedDesktopTask = `-- name: LeaseParkedDesktopTask :one
UPDATE desktop_tasks
SET active_attempt_id = $1,
lease_token_hash = $2,
lease_expires_at = now() + interval '10 minutes',
status = 'in_progress',
attempts = attempts + 1,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = $3
AND target_client_id = $4
AND status = 'waiting_user'
AND active_attempt_id IS NULL
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
`
type LeaseParkedDesktopTaskParams struct {
AttemptID pgtype.UUID `json:"attempt_id"`
LeaseTokenHash []byte `json:"lease_token_hash"`
DesktopID pgtype.UUID `json:"desktop_id"`
ClientID pgtype.UUID `json:"client_id"`
}
func (q *Queries) LeaseParkedDesktopTask(ctx context.Context, arg LeaseParkedDesktopTaskParams) (DesktopTask, error) {
row := q.db.QueryRow(ctx, leaseParkedDesktopTask,
arg.AttemptID,
arg.LeaseTokenHash,
arg.DesktopID,
arg.ClientID,
)
var i DesktopTask
err := row.Scan(
&i.ID,
&i.DesktopID,
&i.JobID,
&i.TenantID,
&i.WorkspaceID,
&i.TargetAccountID,
&i.TargetClientID,
&i.PlatformID,
&i.Kind,
&i.Payload,
&i.Status,
&i.DedupKey,
&i.ActiveAttemptID,
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -607,12 +535,11 @@ SET active_attempt_id = $1,
lease_expires_at = now() + interval '10 minutes',
status = 'in_progress',
attempts = attempts + 1,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = $3
AND target_client_id = $4
AND status = 'queued'
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
`
type LeaseQueuedDesktopTaskByDesktopIDParams struct {
@@ -647,7 +574,6 @@ func (q *Queries) LeaseQueuedDesktopTaskByDesktopID(ctx context.Context, arg Lea
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -663,11 +589,10 @@ SET status = 'unknown',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE target_client_id = $2
AND workspace_id = $3
AND status IN ('waiting_user', 'in_progress')
AND status = 'in_progress'
`
type MarkDesktopTasksUnknownByClientParams struct {
@@ -684,55 +609,6 @@ func (q *Queries) MarkDesktopTasksUnknownByClient(ctx context.Context, arg MarkD
return result.RowsAffected(), nil
}
const parkDesktopTask = `-- name: ParkDesktopTask :one
UPDATE desktop_tasks
SET status = 'waiting_user',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = $1,
updated_at = now()
WHERE desktop_id = $2
AND lease_token_hash = $3
AND status = 'in_progress'
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
`
type ParkDesktopTaskParams struct {
ParkedReason pgtype.Text `json:"parked_reason"`
DesktopID pgtype.UUID `json:"desktop_id"`
LeaseTokenHash []byte `json:"lease_token_hash"`
}
func (q *Queries) ParkDesktopTask(ctx context.Context, arg ParkDesktopTaskParams) (DesktopTask, error) {
row := q.db.QueryRow(ctx, parkDesktopTask, arg.ParkedReason, arg.DesktopID, arg.LeaseTokenHash)
var i DesktopTask
err := row.Scan(
&i.ID,
&i.DesktopID,
&i.JobID,
&i.TenantID,
&i.WorkspaceID,
&i.TargetAccountID,
&i.TargetClientID,
&i.PlatformID,
&i.Kind,
&i.Payload,
&i.Status,
&i.DedupKey,
&i.ActiveAttemptID,
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const reconcileDesktopTask = `-- name: ReconcileDesktopTask :one
UPDATE desktop_tasks
SET status = CASE
@@ -747,13 +623,12 @@ SET status = CASE
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
attempts = attempts + CASE WHEN $1::text = 'retry' THEN 1 ELSE 0 END,
updated_at = now()
WHERE desktop_id = $4
AND workspace_id = $5
AND status = 'unknown'
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
`
type ReconcileDesktopTaskParams struct {
@@ -790,7 +665,6 @@ func (q *Queries) ReconcileDesktopTask(ctx context.Context, arg ReconcileDesktop
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -806,12 +680,11 @@ SET status = 'aborted',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = $2
AND workspace_id = $3
AND status IN ('queued', 'waiting_user')
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, parked_reason, result, error, created_at, updated_at
AND status = 'queued'
RETURNING id, desktop_id, job_id, tenant_id, workspace_id, target_account_id, target_client_id, platform_id, kind, payload, status, dedup_key, active_attempt_id, lease_token_hash, lease_expires_at, attempts, result, error, created_at, updated_at
`
type TenantCancelDesktopTaskParams struct {
@@ -840,7 +713,6 @@ func (q *Queries) TenantCancelDesktopTask(ctx context.Context, arg TenantCancelD
&i.LeaseTokenHash,
&i.LeaseExpiresAt,
&i.Attempts,
&i.ParkedReason,
&i.Result,
&i.Error,
&i.CreatedAt,
@@ -166,7 +166,6 @@ type DesktopTask struct {
LeaseTokenHash []byte `json:"lease_token_hash"`
LeaseExpiresAt pgtype.Timestamptz `json:"lease_expires_at"`
Attempts int32 `json:"attempts"`
ParkedReason pgtype.Text `json:"parked_reason"`
Result []byte `json:"result"`
Error []byte `json:"error"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
@@ -64,6 +64,7 @@ type Querier interface {
GetCurrentBalance(ctx context.Context, arg GetCurrentBalanceParams) (int32, error)
GetDesktopAccountByDesktopID(ctx context.Context, arg GetDesktopAccountByDesktopIDParams) (GetDesktopAccountByDesktopIDRow, error)
GetDesktopAccountByIdentity(ctx context.Context, arg GetDesktopAccountByIdentityParams) (GetDesktopAccountByIdentityRow, error)
GetDesktopClientByID(ctx context.Context, arg GetDesktopClientByIDParams) (DesktopClient, error)
GetDesktopClientByTokenHash(ctx context.Context, tokenHash []byte) (DesktopClient, error)
GetDesktopTaskByDesktopID(ctx context.Context, arg GetDesktopTaskByDesktopIDParams) (DesktopTask, error)
GetImageAsset(ctx context.Context, arg GetImageAssetParams) (GetImageAssetRow, error)
@@ -101,7 +102,6 @@ type Querier interface {
// Cross-tenant aggregation for KOL dashboard.
KolUsageCountByPackage(ctx context.Context, packageIds []int64) ([]KolUsageCountByPackageRow, error)
LeaseNextQueuedDesktopTask(ctx context.Context, arg LeaseNextQueuedDesktopTaskParams) (DesktopTask, error)
LeaseParkedDesktopTask(ctx context.Context, arg LeaseParkedDesktopTaskParams) (DesktopTask, error)
LeaseQueuedDesktopTaskByDesktopID(ctx context.Context, arg LeaseQueuedDesktopTaskByDesktopIDParams) (DesktopTask, error)
ListActiveImagesByFolder(ctx context.Context, arg ListActiveImagesByFolderParams) ([]ListActiveImagesByFolderRow, error)
ListActiveKolPromptsByPackage(ctx context.Context, arg ListActiveKolPromptsByPackageParams) ([]ListActiveKolPromptsByPackageRow, error)
@@ -112,7 +112,7 @@ type Querier interface {
ListArticles(ctx context.Context, arg ListArticlesParams) ([]ListArticlesRow, error)
ListBrands(ctx context.Context, tenantID int64) ([]ListBrandsRow, error)
ListCompetitors(ctx context.Context, arg ListCompetitorsParams) ([]ListCompetitorsRow, error)
ListDesktopAccountsByClient(ctx context.Context, arg ListDesktopAccountsByClientParams) ([]ListDesktopAccountsByClientRow, error)
ListDesktopAccountsByUser(ctx context.Context, arg ListDesktopAccountsByUserParams) ([]ListDesktopAccountsByUserRow, error)
ListDesktopClientsByWorkspace(ctx context.Context, workspaceID int64) ([]DesktopClient, error)
ListImageAssets(ctx context.Context, arg ListImageAssetsParams) ([]ListImageAssetsRow, error)
ListImageFolders(ctx context.Context, tenantID int64) ([]ListImageFoldersRow, error)
@@ -152,7 +152,6 @@ type Querier interface {
MarkKolUsageFailed(ctx context.Context, arg MarkKolUsageFailedParams) error
MarkKolUsageFailedByTask(ctx context.Context, arg MarkKolUsageFailedByTaskParams) error
NextKolPromptRevisionNo(ctx context.Context, arg NextKolPromptRevisionNoParams) (int32, error)
ParkDesktopTask(ctx context.Context, arg ParkDesktopTaskParams) (DesktopTask, error)
PatchDesktopAccount(ctx context.Context, arg PatchDesktopAccountParams) (PatchDesktopAccountRow, error)
ReconcileDesktopTask(ctx context.Context, arg ReconcileDesktopTaskParams) (DesktopTask, error)
RefundReservation(ctx context.Context, arg RefundReservationParams) error
@@ -1,4 +1,4 @@
-- name: ListDesktopAccountsByClient :many
-- name: ListDesktopAccountsByUser :many
SELECT
desktop_id,
tenant_id,
@@ -9,6 +9,7 @@ SELECT
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -19,7 +20,8 @@ SELECT
updated_at
FROM platform_accounts
WHERE workspace_id = sqlc.arg(workspace_id)
AND client_id = sqlc.arg(client_id)
AND user_id = sqlc.arg(user_id)
AND deleted_at IS NULL
ORDER BY platform_id, display_name, created_at DESC;
-- name: GetDesktopAccountByDesktopID :one
@@ -33,6 +35,7 @@ SELECT
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -57,6 +60,7 @@ SELECT
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -82,6 +86,7 @@ INSERT INTO platform_accounts (
platform_id,
platform_uid,
nickname,
avatar_url,
status,
metadata_json,
last_check_at,
@@ -101,6 +106,7 @@ VALUES (
sqlc.arg(platform_id),
sqlc.arg(platform_uid),
sqlc.arg(display_name)::text,
sqlc.narg(avatar_url),
sqlc.arg(legacy_status),
sqlc.narg(metadata_json),
sqlc.narg(verified_at),
@@ -116,6 +122,7 @@ DO UPDATE SET
client_id = EXCLUDED.client_id,
user_id = EXCLUDED.user_id,
nickname = EXCLUDED.nickname,
avatar_url = COALESCE(EXCLUDED.avatar_url, platform_accounts.avatar_url),
status = EXCLUDED.status,
metadata_json = COALESCE(EXCLUDED.metadata_json, platform_accounts.metadata_json),
last_check_at = COALESCE(EXCLUDED.last_check_at, platform_accounts.last_check_at),
@@ -142,6 +149,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -176,6 +184,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -204,6 +213,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -231,6 +241,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -258,6 +269,7 @@ RETURNING
platform_uid,
account_fingerprint,
display_name,
avatar_url,
health,
verified_at,
tags,
@@ -23,6 +23,19 @@ VALUES (
sqlc.narg(client_version),
sqlc.narg(channel)
)
ON CONFLICT (id) DO UPDATE
SET tenant_id = EXCLUDED.tenant_id,
workspace_id = EXCLUDED.workspace_id,
user_id = EXCLUDED.user_id,
token_hash = EXCLUDED.token_hash,
device_name = EXCLUDED.device_name,
os = EXCLUDED.os,
cpu_arch = EXCLUDED.cpu_arch,
client_version = EXCLUDED.client_version,
channel = EXCLUDED.channel,
last_seen_at = now(),
last_rotated_at = now(),
revoked_at = NULL
RETURNING *;
-- name: GetDesktopClientByTokenHash :one
@@ -32,6 +45,14 @@ WHERE token_hash = sqlc.arg(token_hash)
AND revoked_at IS NULL
LIMIT 1;
-- name: GetDesktopClientByID :one
SELECT *
FROM desktop_clients
WHERE id = sqlc.arg(id)
AND workspace_id = sqlc.arg(workspace_id)
AND revoked_at IS NULL
LIMIT 1;
-- name: RotateDesktopClientToken :one
UPDATE desktop_clients
SET token_hash = sqlc.arg(token_hash),
@@ -31,8 +31,7 @@ INSERT INTO desktop_tasks (
kind,
payload,
status,
dedup_key,
parked_reason
dedup_key
)
VALUES (
sqlc.arg(desktop_id),
@@ -45,8 +44,7 @@ VALUES (
sqlc.arg(kind),
sqlc.arg(payload),
sqlc.arg(status),
sqlc.narg(dedup_key),
sqlc.narg(parked_reason)
sqlc.narg(dedup_key)
)
RETURNING *;
@@ -77,7 +75,6 @@ SET active_attempt_id = sqlc.arg(attempt_id),
lease_expires_at = now() + interval '10 minutes',
status = 'in_progress',
attempts = t.attempts + 1,
parked_reason = NULL,
updated_at = now()
FROM candidate
WHERE t.desktop_id = candidate.desktop_id
@@ -90,28 +87,12 @@ SET active_attempt_id = sqlc.arg(attempt_id),
lease_expires_at = now() + interval '10 minutes',
status = 'in_progress',
attempts = attempts + 1,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id)
AND target_client_id = sqlc.arg(client_id)
AND status = 'queued'
RETURNING *;
-- name: LeaseParkedDesktopTask :one
UPDATE desktop_tasks
SET active_attempt_id = sqlc.arg(attempt_id),
lease_token_hash = sqlc.arg(lease_token_hash),
lease_expires_at = now() + interval '10 minutes',
status = 'in_progress',
attempts = attempts + 1,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id)
AND target_client_id = sqlc.arg(client_id)
AND status = 'waiting_user'
AND active_attempt_id IS NULL
RETURNING *;
-- name: ExtendDesktopTaskLease :one
UPDATE desktop_tasks
SET lease_expires_at = now() + interval '10 minutes',
@@ -121,19 +102,6 @@ WHERE desktop_id = sqlc.arg(desktop_id)
AND status = 'in_progress'
RETURNING *;
-- name: ParkDesktopTask :one
UPDATE desktop_tasks
SET status = 'waiting_user',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = sqlc.arg(parked_reason),
updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id)
AND lease_token_hash = sqlc.arg(lease_token_hash)
AND status = 'in_progress'
RETURNING *;
-- name: CompleteDesktopTask :one
UPDATE desktop_tasks
SET status = sqlc.arg(status),
@@ -142,7 +110,6 @@ SET status = sqlc.arg(status),
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id)
AND lease_token_hash = sqlc.arg(lease_token_hash)
@@ -156,7 +123,6 @@ SET status = 'aborted',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id)
AND lease_token_hash = sqlc.arg(lease_token_hash)
@@ -170,11 +136,10 @@ SET status = 'aborted',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id)
AND target_client_id = sqlc.arg(client_id)
AND status IN ('queued', 'waiting_user')
AND status = 'queued'
RETURNING *;
-- name: TenantCancelDesktopTask :one
@@ -184,11 +149,10 @@ SET status = 'aborted',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id)
AND workspace_id = sqlc.arg(workspace_id)
AND status IN ('queued', 'waiting_user')
AND status = 'queued'
RETURNING *;
-- name: ReconcileDesktopTask :one
@@ -205,7 +169,6 @@ SET status = CASE
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
attempts = attempts + CASE WHEN sqlc.arg(status)::text = 'retry' THEN 1 ELSE 0 END,
updated_at = now()
WHERE desktop_id = sqlc.arg(desktop_id)
@@ -220,11 +183,10 @@ SET status = 'unknown',
active_attempt_id = NULL,
lease_token_hash = NULL,
lease_expires_at = NULL,
parked_reason = NULL,
updated_at = now()
WHERE target_client_id = sqlc.arg(client_id)
AND workspace_id = sqlc.arg(workspace_id)
AND status IN ('waiting_user', 'in_progress');
AND status = 'in_progress';
-- name: CreateDesktopTaskAttempt :one
INSERT INTO desktop_task_attempts (
@@ -21,6 +21,8 @@ func NewDesktopAccountHandler(a *bootstrap.App) *DesktopAccountHandler {
return &DesktopAccountHandler{
svc: app.NewDesktopAccountService(
repository.NewDesktopAccountRepository(a.DB),
repository.NewDesktopClientRepository(a.DB),
a.Redis,
),
}
}
@@ -31,7 +33,16 @@ func (h *DesktopAccountHandler) List(c *gin.Context) {
return
}
data, err := h.svc.ListByClient(c.Request.Context(), MustDesktopClient(c.Request.Context()))
data, err := h.svc.ListByUser(c.Request.Context(), MustDesktopClient(c.Request.Context()))
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopAccountHandler) TenantList(c *gin.Context) {
data, err := h.svc.ListForActor(c.Request.Context(), auth.MustActor(c.Request.Context()))
if err != nil {
response.Error(c, err)
return
@@ -22,6 +22,10 @@ func (s *stubDesktopClientRepo) Register(ctx context.Context, params repository.
panic("unexpected call")
}
func (s *stubDesktopClientRepo) GetByID(ctx context.Context, id uuid.UUID, workspaceID int64) (*repository.DesktopClient, error) {
panic("unexpected call")
}
func (s *stubDesktopClientRepo) GetByTokenHash(ctx context.Context, tokenHash []byte) (*repository.DesktopClient, error) {
if s.client == nil {
return nil, context.Canceled
@@ -18,6 +18,7 @@ func NewDesktopClientHandler(a *bootstrap.App) *DesktopClientHandler {
return &DesktopClientHandler{
svc: app.NewDesktopClientService(
repository.NewDesktopClientRepository(a.DB),
a.Redis,
),
}
}
@@ -71,3 +72,12 @@ func (h *DesktopClientHandler) Revoke(c *gin.Context) {
}
response.Success(c, data)
}
func (h *DesktopClientHandler) Offline(c *gin.Context) {
data, err := h.svc.Offline(c.Request.Context(), MustDesktopClient(c.Request.Context()))
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
@@ -3,6 +3,8 @@ package transport
import (
"errors"
"io"
"strconv"
"strings"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
@@ -11,20 +13,26 @@ import (
"github.com/geo-platform/tenant-api/internal/shared/auth"
"github.com/geo-platform/tenant-api/internal/shared/response"
"github.com/geo-platform/tenant-api/internal/tenant/app"
"github.com/geo-platform/tenant-api/internal/tenant/repository"
)
type DesktopTaskHandler struct {
svc *app.DesktopTaskService
svc *app.DesktopTaskService
publishSvc *app.PublishJobService
}
func NewDesktopTaskHandler(a *bootstrap.App) *DesktopTaskHandler {
return &DesktopTaskHandler{
svc: app.NewDesktopTaskService(
repository.NewDesktopTaskRepository(a.DB),
a.DB,
a.RabbitMQ,
a.Logger,
),
).WithCache(a.Cache),
publishSvc: app.NewPublishJobService(
a.DB,
a.RabbitMQ,
a.Redis,
a.Logger,
).WithCache(a.Cache),
}
}
@@ -45,7 +53,44 @@ func (h *DesktopTaskHandler) Lease(c *gin.Context) {
routeTaskID = &parsed
}
data, err := h.svc.Lease(c.Request.Context(), MustDesktopClient(c.Request.Context()), req, routeTaskID, c.Query("from_parked") == "true")
data, err := h.svc.Lease(c.Request.Context(), MustDesktopClient(c.Request.Context()), req, routeTaskID)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) ListPublish(c *gin.Context) {
req := app.ListPublishTasksRequest{
Page: 1,
PageSize: 10,
Title: strings.TrimSpace(c.Query("title")),
}
if rawPage := c.Query("page"); rawPage != "" {
parsed, err := strconv.Atoi(rawPage)
if err != nil || parsed <= 0 {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", "page must be a positive integer"))
return
}
req.Page = parsed
}
rawPageSize := c.Query("page_size")
if rawPageSize == "" {
rawPageSize = c.Query("limit")
}
if rawPageSize != "" {
parsed, err := strconv.Atoi(rawPageSize)
if err != nil || parsed <= 0 {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", "page_size must be a positive integer"))
return
}
req.PageSize = parsed
}
data, err := h.svc.ListPublishTasks(c.Request.Context(), MustDesktopClient(c.Request.Context()), req)
if err != nil {
response.Error(c, err)
return
@@ -74,27 +119,6 @@ func (h *DesktopTaskHandler) Extend(c *gin.Context) {
response.Success(c, data)
}
func (h *DesktopTaskHandler) Park(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
var req app.ParkDesktopTaskRequest
if err := c.ShouldBindJSON(&req); err != nil {
response.Error(c, response.ErrBadRequest(40001, "invalid_params", err.Error()))
return
}
data, err := h.svc.Park(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID, req)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) Result(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
@@ -116,6 +140,21 @@ func (h *DesktopTaskHandler) Result(c *gin.Context) {
response.Success(c, data)
}
func (h *DesktopTaskHandler) RetryPublish(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
response.Error(c, response.ErrBadRequest(40084, "invalid_desktop_task_id", "task id must be a uuid"))
return
}
data, err := h.publishSvc.RetryByDesktopTask(c.Request.Context(), MustDesktopClient(c.Request.Context()), desktopID)
if err != nil {
response.Error(c, err)
return
}
response.Success(c, data)
}
func (h *DesktopTaskHandler) Cancel(c *gin.Context) {
desktopID, err := uuid.Parse(c.Param("id"))
if err != nil {
@@ -15,7 +15,7 @@ type PublishJobHandler struct {
func NewPublishJobHandler(a *bootstrap.App) *PublishJobHandler {
return &PublishJobHandler{
svc: app.NewPublishJobService(a.DB, a.RabbitMQ, a.Logger),
svc: app.NewPublishJobService(a.DB, a.RabbitMQ, a.Redis, a.Logger).WithCache(a.Cache),
}
}
+7 -1
View File
@@ -40,27 +40,33 @@ func RegisterRoutes(app *bootstrap.App) {
desktopAuth.Use(DesktopClientMiddleware(repository.NewDesktopClientRepository(app.DB)))
desktopAuth.POST("/clients/rotate", desktopClientHandler.Rotate)
desktopAuth.POST("/clients/heartbeat", desktopClientHandler.Heartbeat)
desktopAuth.POST("/clients/offline", desktopClientHandler.Offline)
desktopAuth.POST("/clients/revoke", desktopClientHandler.Revoke)
desktopAuth.GET("/events", desktopEventsHandler.Stream)
desktopAuth.GET("/accounts", desktopAccountHandler.List)
desktopAuth.GET("/content/articles/:id", desktopContentHandler.Article)
desktopAuth.GET("/publish-tasks", desktopTaskHandler.ListPublish)
desktopAuth.POST("/accounts", desktopAccountHandler.Upsert)
desktopAuth.PATCH("/accounts/:id", desktopAccountHandler.Patch)
desktopAuth.DELETE("/accounts/:id", desktopAccountHandler.Delete)
desktopAuth.POST("/tasks/lease", desktopTaskHandler.Lease)
desktopAuth.POST("/tasks/:id/lease", desktopTaskHandler.Lease)
desktopAuth.POST("/tasks/:id/extend", desktopTaskHandler.Extend)
desktopAuth.POST("/tasks/:id/park", desktopTaskHandler.Park)
desktopAuth.POST("/tasks/:id/cancel", desktopTaskHandler.Cancel)
desktopAuth.POST("/tasks/:id/result", desktopTaskHandler.Result)
desktopAuth.POST("/publish-tasks/:id/retry", desktopTaskHandler.RetryPublish)
tenantProtected := protected.Group("/tenant")
tenantProtected.Use(RequireActiveTenantSubscription(repository.NewTenantPlanRepository(app.DB)))
tenantProtected.GET("/accounts", desktopAccountHandler.TenantList)
tenantProtected.POST("/accounts/:id/request-delete", desktopAccountHandler.RequestDelete)
tenantProtected.POST("/tasks/:id/reconcile", desktopTaskHandler.Reconcile)
tenantProtected.POST("/tasks/:id/cancel", desktopTaskHandler.TenantCancel)
tenantProtected.POST("/publish-jobs", publishJobHandler.Create)
media := tenantProtected.Group("/media")
media.GET("/platforms", pluginHandler.ListPlatforms)
workspace := tenantProtected.Group("/workspace")
wsHandler := NewWorkspaceHandler(app)
workspace.GET("/overview", wsHandler.Overview)
@@ -26,10 +26,12 @@ func TestProtectedRoutes_RequireAuth(t *testing.T) {
{http.MethodGet, "/api/auth/me"},
{http.MethodPost, "/api/auth/logout"},
{http.MethodPost, "/api/desktop/clients/register"},
{http.MethodGet, "/api/tenant/accounts"},
{http.MethodPost, "/api/tenant/accounts/:id/request-delete"},
{http.MethodPost, "/api/tenant/tasks/:id/reconcile"},
{http.MethodPost, "/api/tenant/tasks/:id/cancel"},
{http.MethodPost, "/api/tenant/publish-jobs"},
{http.MethodGet, "/api/tenant/media/platforms"},
{http.MethodGet, "/api/tenant/workspace/overview"},
{http.MethodGet, "/api/tenant/templates"},
{http.MethodGet, "/api/tenant/articles"},
@@ -75,6 +77,7 @@ func TestDesktopClientRoutes_ClientTokenAuth(t *testing.T) {
}{
{http.MethodGet, "/api/desktop/events"},
{http.MethodGet, "/api/desktop/accounts"},
{http.MethodPost, "/api/desktop/clients/offline"},
{http.MethodPost, "/api/desktop/tasks/lease"},
{http.MethodPost, "/api/desktop/tasks/:id/result"},
}