root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(tenant): drop legacy plugin_installations, migrate monitoring to desktop_clients

Browse Source 
Hard cutover from the browser-extension plugin flow to desktop clients:
remove plugin_installations/plugin_sessions tables and related service,
handler, router, and generated model code; migrate monitoring quotas
and collector types to desktop_clients (UUID primary_client_id);
recreate platform_access_snapshots keyed by client_id; update dev-seed
and callback types accordingly; mark legacy design docs as historical.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
liangxu
2026-04-20 13:52:35 +08:00
parent f5a1cd81b2
commit 5ff2e2e74c
25 changed files with 535 additions and 1384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/monitoring_callback_service.go
+63 -62
View File
@@ -100,7 +100,7 @@ type MonitoringHeartbeatPlatform struct {
}
type MonitoringHeartbeatResponse struct {
InstallationID int64 `json:"installation_id"`
InstallationID string `json:"installation_id"`
BusinessDate string `json:"business_date"`
UpdatedPlatformCount int `json:"updated_platform_count"`
ReconciledSkippedTaskCount int64 `json:"reconciled_skipped_task_count"`
@@ -166,8 +166,9 @@ type MonitoringSkipTaskResponse struct {
}
type monitoringInstallationIdentity struct {
ID int64
TenantID int64
ID string
TenantID int64
WorkspaceID int64
}
type monitoringPlatformAccessSnapshotState struct {
@@ -236,7 +237,7 @@ type monitoringTaskResultEnvelope struct {
Version string `json:"version"`
TaskID int64 `json:"task_id"`
TenantID int64 `json:"tenant_id"`
InstallationID int64 `json:"installation_id"`
InstallationID string `json:"installation_id"`
ReceivedAt time.Time `json:"received_at"`
QueueStatus string `json:"queue_status"`
QueueClaimedAt *time.Time `json:"queue_claimed_at"`
@@ -256,7 +257,7 @@ type monitoringTaskResultQueueMetaPatch struct {
type MonitoringTaskResultQueueMetaPatch = monitoringTaskResultQueueMetaPatch
func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID int64, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) {
func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringTaskResultRequest) (*MonitoringTaskResultResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
@@ -268,10 +269,10 @@ func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, instal
}
if task.CollectorType != monitoringCollectorType {
return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not plugin")
return nil, response.ErrConflict(40941, "collector_type_mismatch", "task collector type is not desktop")
}
if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != strconv.FormatInt(installation.ID, 10) {
if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != installation.ID {
return nil, response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation")
}
@@ -356,13 +357,13 @@ func (s *MonitoringCallbackService) HandleTaskResult(ctx context.Context, instal
}, nil
}
func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID int64, installationToken string, req MonitoringHeartbeatRequest) (*MonitoringHeartbeatResponse, error) {
func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationID string, installationToken string, req MonitoringHeartbeatRequest) (*MonitoringHeartbeatResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
}
isPrimaryInstallation, err := s.resolveHeartbeatPrimaryInstallation(ctx, installation.TenantID, installation.ID)
isPrimaryInstallation, err := s.resolveHeartbeatPrimaryInstallation(ctx, installation.TenantID, installation)
if err != nil {
return nil, err
}
@@ -442,7 +443,7 @@ func (s *MonitoringCallbackService) Heartbeat(ctx context.Context, installationI
}, nil
}
func (s *MonitoringCallbackService) LeaseTasks(ctx context.Context, installationID int64, installationToken string, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) {
func (s *MonitoringCallbackService) LeaseTasks(ctx context.Context, installationID string, installationToken string, req MonitoringLeaseTasksRequest) (*MonitoringLeaseTasksResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
@@ -522,7 +523,7 @@ func (s *MonitoringCallbackService) LeaseTasks(ctx context.Context, installation
}, nil
}
func (s *MonitoringCallbackService) ResumeTasks(ctx context.Context, installationID int64, installationToken string, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) {
func (s *MonitoringCallbackService) ResumeTasks(ctx context.Context, installationID string, installationToken string, req MonitoringResumeTasksRequest) (*MonitoringResumeTasksResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
@@ -601,7 +602,7 @@ func (s *MonitoringCallbackService) ResumeTasks(ctx context.Context, installatio
}, nil
}
func (s *MonitoringCallbackService) SkipTask(ctx context.Context, installationID int64, installationToken string, taskID int64, req MonitoringSkipTaskRequest) (*MonitoringSkipTaskResponse, error) {
func (s *MonitoringCallbackService) SkipTask(ctx context.Context, installationID string, installationToken string, taskID int64, req MonitoringSkipTaskRequest) (*MonitoringSkipTaskResponse, error) {
installation, err := s.authenticateInstallation(ctx, installationID, installationToken)
if err != nil {
return nil, err
@@ -704,48 +705,47 @@ func (s *MonitoringCallbackService) ProcessQueuedTaskResult(ctx context.Context,
return s.processTaskResult(ctx, task, envelope.CallbackResult)
}
func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context, installationID int64, installationToken string) (*monitoringInstallationIdentity, error) {
tokenHash := sharedauth.HashToken(strings.TrimSpace(installationToken))
func (s *MonitoringCallbackService) authenticateInstallation(ctx context.Context, installationID string, installationToken string) (*monitoringInstallationIdentity, error) {
tokenHash := HashDesktopClientToken(strings.TrimSpace(installationToken))
row := &monitoringInstallationIdentity{}
if err := s.businessPool.QueryRow(ctx, `
SELECT id, tenant_id
FROM plugin_installations
WHERE id = $1
AND installation_token_hash = $2
AND status = 'active'
AND deleted_at IS NULL
`, installationID, tokenHash).Scan(&row.ID, &row.TenantID); err != nil {
SELECT id::text, tenant_id, workspace_id
FROM desktop_clients
WHERE id = $1::uuid
AND token_hash = $2
AND revoked_at IS NULL
`, strings.TrimSpace(installationID), tokenHash).Scan(&row.ID, &row.TenantID, &row.WorkspaceID); err != nil {
if err == pgx.ErrNoRows {
return nil, response.ErrUnauthorized(40141, "plugin_installation_unauthorized", "plugin installation is invalid or expired")
return nil, response.ErrUnauthorized(40141, "desktop_client_unauthorized", "desktop client is invalid or expired")
}
return nil, response.ErrInternal(50041, "plugin_installation_lookup_failed", "failed to validate plugin installation")
return nil, response.ErrInternal(50041, "desktop_client_lookup_failed", "failed to validate desktop client")
}
return row, nil
}
func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx context.Context, tenantID, installationID int64) (bool, error) {
var preferred sql.NullInt64
func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx context.Context, tenantID int64, installation *monitoringInstallationIdentity) (bool, error) {
var preferred sql.NullString
err := s.businessPool.QueryRow(ctx, `
SELECT primary_installation_id
SELECT primary_client_id::text
FROM tenant_monitoring_quotas
WHERE tenant_id = $1
`, tenantID).Scan(&preferred)
if err != nil && err != pgx.ErrNoRows {
return false, response.ErrInternal(50041, "query_failed", "failed to inspect primary monitoring installation")
return false, response.ErrInternal(50041, "query_failed", "failed to inspect primary monitoring desktop client")
}
if !preferred.Valid || preferred.Int64 == installationID {
if !preferred.Valid || preferred.String == installation.ID {
if !preferred.Valid {
if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installationID); err != nil {
if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installation); err != nil {
return false, err
}
}
return true, nil
}
online, err := s.isMonitoringInstallationRecentlyOnline(ctx, tenantID, preferred.Int64)
online, err := s.isMonitoringInstallationRecentlyOnline(ctx, tenantID, installation.WorkspaceID, preferred.String)
if err != nil {
return false, err
}
@@ -753,27 +753,27 @@ func (s *MonitoringCallbackService) resolveHeartbeatPrimaryInstallation(ctx cont
return false, nil
}
if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installationID); err != nil {
if err := s.persistHeartbeatPrimaryInstallationID(ctx, tenantID, installation); err != nil {
return false, err
}
return true, nil
}
func (s *MonitoringCallbackService) isMonitoringInstallationRecentlyOnline(ctx context.Context, tenantID, installationID int64) (bool, error) {
func (s *MonitoringCallbackService) isMonitoringInstallationRecentlyOnline(ctx context.Context, tenantID, workspaceID int64, installationID string) (bool, error) {
var lastSeenAt sql.NullTime
err := s.businessPool.QueryRow(ctx, `
SELECT last_seen_at
FROM plugin_installations
WHERE id = $1
FROM desktop_clients
WHERE id = $1::uuid
AND tenant_id = $2
AND status = 'active'
AND deleted_at IS NULL
`, installationID, tenantID).Scan(&lastSeenAt)
AND workspace_id = $3
AND revoked_at IS NULL
`, strings.TrimSpace(installationID), tenantID, workspaceID).Scan(&lastSeenAt)
if err != nil {
if err == pgx.ErrNoRows {
return false, nil
}
return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring installation heartbeat")
return false, response.ErrInternal(50041, "query_failed", "failed to inspect monitoring desktop client heartbeat")
}
if !lastSeenAt.Valid {
@@ -782,22 +782,22 @@ func (s *MonitoringCallbackService) isMonitoringInstallationRecentlyOnline(ctx c
return time.Since(lastSeenAt.Time) <= monitoringOnlineDuration, nil
}
func (s *MonitoringCallbackService) persistHeartbeatPrimaryInstallationID(ctx context.Context, tenantID, installationID int64) error {
func (s *MonitoringCallbackService) persistHeartbeatPrimaryInstallationID(ctx context.Context, tenantID int64, installation *monitoringInstallationIdentity) error {
if _, err := s.businessPool.Exec(ctx, `
INSERT INTO tenant_monitoring_quotas (
tenant_id, workspace_id, primary_installation_id
tenant_id, workspace_id, primary_client_id
)
VALUES (
$1,
(SELECT id FROM workspaces WHERE tenant_id = $1 AND is_default = TRUE LIMIT 1),
$2
$2,
$3::uuid
)
ON CONFLICT (tenant_id) DO UPDATE
SET workspace_id = COALESCE(tenant_monitoring_quotas.workspace_id, EXCLUDED.workspace_id),
primary_installation_id = EXCLUDED.primary_installation_id,
SET workspace_id = EXCLUDED.workspace_id,
primary_client_id = EXCLUDED.primary_client_id,
updated_at = NOW()
`, tenantID, installationID); err != nil {
return response.ErrInternal(50041, "update_failed", "failed to persist primary monitoring installation")
`, tenantID, installation.WorkspaceID, installation.ID); err != nil {
return response.ErrInternal(50041, "update_failed", "failed to persist primary monitoring desktop client")
}
return nil
}
@@ -1020,11 +1020,12 @@ func (s *MonitoringCallbackService) selectPendingLeaseTasks(
func (s *MonitoringCallbackService) selectLeasedTasksForInstallation(
ctx context.Context,
tx pgx.Tx,
tenantID, installationID int64,
tenantID int64,
installationID string,
businessDate string,
eligiblePlatformIDs []string,
) ([]monitoringLeasedTaskRow, error) {
queryArgs := []interface{}{tenantID, monitoringCollectorType, businessDate, strconv.FormatInt(installationID, 10)}
queryArgs := []interface{}{tenantID, monitoringCollectorType, businessDate, installationID}
query := `
SELECT
t.id,
@@ -1108,7 +1109,7 @@ func (s *MonitoringCallbackService) selectLeasedTasksForInstallation(
return result, nil
}
func (s *MonitoringCallbackService) markTaskLeased(ctx context.Context, tx pgx.Tx, installationID, taskID int64, leaseToken string, leasedAt, leaseExpiresAt time.Time) error {
func (s *MonitoringCallbackService) markTaskLeased(ctx context.Context, tx pgx.Tx, installationID string, taskID int64, leaseToken string, leasedAt, leaseExpiresAt time.Time) error {
if _, err := tx.Exec(ctx, `
UPDATE monitoring_collect_tasks
SET status = 'leased',
@@ -1120,7 +1121,7 @@ func (s *MonitoringCallbackService) markTaskLeased(ctx context.Context, tx pgx.T
error_message = NULL,
updated_at = NOW()
WHERE id = $1
`, taskID, sharedauth.HashToken(strings.TrimSpace(leaseToken)), strconv.FormatInt(installationID, 10), leasedAt, leaseExpiresAt); err != nil {
`, taskID, sharedauth.HashToken(strings.TrimSpace(leaseToken)), installationID, leasedAt, leaseExpiresAt); err != nil {
return response.ErrInternal(50041, "lease_update_failed", "failed to update monitoring lease task")
}
return nil
@@ -1165,11 +1166,11 @@ func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Con
if _, err := tx.Exec(ctx, `
INSERT INTO monitoring_platform_access_snapshots (
tenant_id, installation_id, ai_platform_id, business_date, access_status,
tenant_id, client_id, ai_platform_id, business_date, access_status,
connected, accessible, detected_at, reason_code, reason_message
)
VALUES ($1, $2, $3, $4::date, $5, $6, $7, $8, $9, $10)
ON CONFLICT (tenant_id, installation_id, ai_platform_id, business_date)
VALUES ($1, $2::uuid, $3, $4::date, $5, $6, $7, $8, $9, $10)
ON CONFLICT (tenant_id, client_id, ai_platform_id, business_date)
DO UPDATE SET
access_status = EXCLUDED.access_status,
connected = EXCLUDED.connected,
@@ -1195,7 +1196,7 @@ func (s *MonitoringCallbackService) upsertPlatformAccessSnapshot(ctx context.Con
return hasPlatformAccessSnapshotChanged(previous, nextState), nil
}
func (s *MonitoringCallbackService) loadPlatformAccessSnapshotState(ctx context.Context, tx pgx.Tx, tenantID, installationID int64, businessDate, aiPlatformID string) (*monitoringPlatformAccessSnapshotState, error) {
func (s *MonitoringCallbackService) loadPlatformAccessSnapshotState(ctx context.Context, tx pgx.Tx, tenantID int64, installationID string, businessDate, aiPlatformID string) (*monitoringPlatformAccessSnapshotState, error) {
item := &monitoringPlatformAccessSnapshotState{}
var reasonCode sql.NullString
var reasonMessage sql.NullString
@@ -1203,7 +1204,7 @@ func (s *MonitoringCallbackService) loadPlatformAccessSnapshotState(ctx context.
SELECT access_status, connected, accessible, reason_code, reason_message
FROM monitoring_platform_access_snapshots
WHERE tenant_id = $1
AND installation_id = $2
AND client_id = $2::uuid
AND ai_platform_id = $3
AND business_date = $4::date
`, tenantID, installationID, aiPlatformID, businessDate).Scan(
@@ -1721,11 +1722,11 @@ func (s *MonitoringCallbackService) loadExistingRunSummary(ctx context.Context,
return &value, citationSourceCount, contentCitationCount, nil
}
func (s *MonitoringCallbackService) touchInstallation(ctx context.Context, installationID int64) error {
func (s *MonitoringCallbackService) touchInstallation(ctx context.Context, installationID string) error {
_, err := s.businessPool.Exec(ctx, `
UPDATE plugin_installations
SET last_seen_at = NOW(), updated_at = NOW()
WHERE id = $1
UPDATE desktop_clients
SET last_seen_at = NOW()
WHERE id = $1::uuid
`, installationID)
return err
}
@@ -1821,7 +1822,7 @@ func buildMonitoringRawPayload(req MonitoringTaskResultRequest) ([]byte, []Monit
return data, sourceInputs
}
func newMonitoringTaskResultEnvelope(task *monitoringCollectTask, installationID int64, receivedAt time.Time, req MonitoringTaskResultRequest) monitoringTaskResultEnvelope {
func newMonitoringTaskResultEnvelope(task *monitoringCollectTask, installationID string, receivedAt time.Time, req MonitoringTaskResultRequest) monitoringTaskResultEnvelope {
return monitoringTaskResultEnvelope{
Version: "v1",
TaskID: task.ID,
@@ -2090,8 +2091,8 @@ func normalizeMonitoringAccessStatus(accessStatus *string, connected *bool, acce
return "", false, false, response.ErrBadRequest(40041, "invalid_access_status", "access status requires access_status, connected, or accessible")
}
func validateTaskLease(task *monitoringCollectTask, installationID int64, leaseToken string) error {
if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != strconv.FormatInt(installationID, 10) {
func validateTaskLease(task *monitoringCollectTask, installationID string, leaseToken string) error {
if task.LeasedToExecutor.Valid && strings.TrimSpace(task.LeasedToExecutor.String) != "" && task.LeasedToExecutor.String != installationID {
return response.ErrUnauthorized(40141, "lease_owner_mismatch", "task is leased to another installation")
}