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

feat(cache): add read-through cache layer across all app services

Browse Source 
Introduce a generic read-through caching infrastructure and wire it into
all major tenant app services to reduce database load on hot read paths.

Key changes:
- Add `DeletePrefix` to Cache interface with memory (prefix scan) and
  Redis (SCAN + DEL) implementations
- New `readthrough.go`: generic `LoadJSON` / `LoadJSONWithEmpty` helpers
  backed by singleflight to prevent cache stampedes; supports jittered TTL
- New `cache_support.go`: centralized cache key builders and invalidation
  helpers for all entities (workspace, brand, prompt rules, schedule tasks,
  articles)
- Wire optional cache into ArticleService, BrandService, WorkspaceService,
  PromptRuleService, ScheduleTaskService, TemplateService, MediaService,
  PromptGenerateService via `WithCache()` builder pattern
- ScheduleDispatchWorker invalidates schedule task cache after dispatching
- ArticleService gains a new `Detail` endpoint with empty-result caching
- Update cmd entrypoints and transport handlers to propagate cache
This commit is contained in:
liangxu
2026-04-15 16:11:05 +08:00
parent 4d06938565
commit 1538a12042
28 changed files with 1316 additions and 634 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/internal/tenant/app/schedule_task_service.go
+140 -111
View File
@@ -3,28 +3,39 @@ package app
import (
"context"
"encoding/json"
"errors"
"fmt"
"time"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/jackc/pgx/v5/pgxpool"
"golang.org/x/sync/singleflight"
"github.com/geo-platform/tenant-api/internal/shared/auditlog"
"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/middleware"
"github.com/geo-platform/tenant-api/internal/shared/response"
sharedschedule "github.com/geo-platform/tenant-api/internal/shared/schedule"
)
type ScheduleTaskService struct {
pool *pgxpool.Pool
auditLogs *auditlog.AsyncWriter
pool *pgxpool.Pool
auditLogs *auditlog.AsyncWriter
cache sharedcache.Cache
cacheGroup singleflight.Group
}
func NewScheduleTaskService(pool *pgxpool.Pool, auditLogs *auditlog.AsyncWriter) *ScheduleTaskService {
return &ScheduleTaskService{pool: pool, auditLogs: auditLogs}
}
func (s *ScheduleTaskService) WithCache(c sharedcache.Cache) *ScheduleTaskService {
s.cache = c
return s
}
type ScheduleTaskRequest struct {
PromptRuleID int64 `json:"prompt_rule_id" binding:"required"`
BrandID *int64 `json:"brand_id"`
@@ -77,122 +88,23 @@ type ScheduleTaskStatusRequest struct {
func (s *ScheduleTaskService) List(ctx context.Context, params ScheduleTaskListParams) (*ScheduleTaskListResponse, error) {
actor := auth.MustActor(ctx)
if params.Page < 1 {
params.Page = 1
}
if params.PageSize < 1 || params.PageSize > 100 {
params.PageSize = 20
}
offset := (params.Page - 1) * params.PageSize
var total int64
err := s.pool.QueryRow(ctx, `
SELECT COUNT(*) FROM schedule_tasks
WHERE tenant_id = $1 AND deleted_at IS NULL
AND ($2::bigint IS NULL OR prompt_rule_id = $2)
AND ($3::text IS NULL OR status = $3)
AND ($4::text IS NULL OR name ILIKE '%' || $4 || '%')
AND ($5::timestamptz IS NULL OR created_at >= $5)
AND ($6::timestamptz IS NULL OR created_at <= $6)
`, actor.TenantID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo).Scan(&total)
if err != nil {
return nil, response.ErrInternal(50010, "count_failed", "failed to count schedule tasks")
}
rows, err := s.pool.Query(ctx, `
SELECT st.id, st.prompt_rule_id, st.brand_id, st.name,
st.cron_expr, st.target_platform, st.enable_web_search, st.generate_count, st.start_at, st.end_at,
st.next_run_at, st.status, st.created_at, st.updated_at,
pr.name AS prompt_rule_name,
b.name AS brand_name
FROM schedule_tasks st
LEFT JOIN prompt_rules pr ON pr.id = st.prompt_rule_id
LEFT JOIN brands b ON b.id = st.brand_id
WHERE st.tenant_id = $1 AND st.deleted_at IS NULL
AND ($2::bigint IS NULL OR st.prompt_rule_id = $2)
AND ($3::text IS NULL OR st.status = $3)
AND ($4::text IS NULL OR st.name ILIKE '%' || $4 || '%')
AND ($5::timestamptz IS NULL OR st.created_at >= $5)
AND ($6::timestamptz IS NULL OR st.created_at <= $6)
ORDER BY st.created_at DESC
LIMIT $7 OFFSET $8
`, actor.TenantID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo, params.PageSize, offset)
if err != nil {
return nil, response.ErrInternal(50010, "query_failed", "failed to list schedule tasks")
}
defer rows.Close()
var items []ScheduleTaskResponse
for rows.Next() {
var r ScheduleTaskResponse
var ca, ua interface{}
var startAt, endAt, nextRunAt interface{}
if err := rows.Scan(&r.ID, &r.PromptRuleID, &r.BrandID, &r.Name,
&r.CronExpr, &r.TargetPlatform, &r.EnableWebSearch, &r.GenerateCount, &startAt, &endAt,
&nextRunAt, &r.Status, &ca, &ua,
&r.PromptRuleName, &r.BrandName); err != nil {
return nil, response.ErrInternal(50010, "scan_failed", err.Error())
}
r.CreatedAt = fmt.Sprintf("%v", ca)
r.UpdatedAt = fmt.Sprintf("%v", ua)
if startAt != nil {
s := fmt.Sprintf("%v", startAt)
r.StartAt = &s
}
if endAt != nil {
s := fmt.Sprintf("%v", endAt)
r.EndAt = &s
}
if nextRunAt != nil {
s := fmt.Sprintf("%v", nextRunAt)
r.NextRunAt = &s
}
items = append(items, r)
}
if items == nil {
items = []ScheduleTaskResponse{}
}
return &ScheduleTaskListResponse{Items: items, Total: total}, nil
return sharedcache.LoadJSON(ctx, s.cache, &s.cacheGroup, scheduleTaskListCacheKey(actor.TenantID, params), defaultCacheTTL(), func(loadCtx context.Context) (*ScheduleTaskListResponse, error) {
return s.loadScheduleTasks(loadCtx, actor.TenantID, params)
})
}
func (s *ScheduleTaskService) Detail(ctx context.Context, id int64) (*ScheduleTaskResponse, error) {
actor := auth.MustActor(ctx)
var r ScheduleTaskResponse
var ca, ua interface{}
var startAt, endAt, nextRunAt interface{}
err := s.pool.QueryRow(ctx, `
SELECT st.id, st.prompt_rule_id, st.brand_id, st.name,
st.cron_expr, st.target_platform, st.enable_web_search, st.generate_count, st.start_at, st.end_at,
st.next_run_at, st.status, st.created_at, st.updated_at,
pr.name AS prompt_rule_name,
b.name AS brand_name
FROM schedule_tasks st
LEFT JOIN prompt_rules pr ON pr.id = st.prompt_rule_id
LEFT JOIN brands b ON b.id = st.brand_id
WHERE st.id = $1 AND st.tenant_id = $2 AND st.deleted_at IS NULL
`, id, actor.TenantID).Scan(&r.ID, &r.PromptRuleID, &r.BrandID, &r.Name,
&r.CronExpr, &r.TargetPlatform, &r.EnableWebSearch, &r.GenerateCount, &startAt, &endAt,
&nextRunAt, &r.Status, &ca, &ua,
&r.PromptRuleName, &r.BrandName)
record, found, err := sharedcache.LoadJSONWithEmpty(ctx, s.cache, &s.cacheGroup, scheduleTaskDetailCacheKey(actor.TenantID, id), defaultCacheTTL(), defaultCacheEmptyTTL(), func(loadCtx context.Context) (*ScheduleTaskResponse, bool, error) {
return s.loadScheduleTaskDetail(loadCtx, actor.TenantID, id)
})
if err != nil {
return nil, err
}
if !found || record == nil {
return nil, response.ErrNotFound(40432, "schedule_not_found", "schedule task not found")
}
r.CreatedAt = fmt.Sprintf("%v", ca)
r.UpdatedAt = fmt.Sprintf("%v", ua)
if startAt != nil {
sv := fmt.Sprintf("%v", startAt)
r.StartAt = &sv
}
if endAt != nil {
sv := fmt.Sprintf("%v", endAt)
r.EndAt = &sv
}
if nextRunAt != nil {
sv := fmt.Sprintf("%v", nextRunAt)
r.NextRunAt = &sv
}
return &r, nil
return record, nil
}
func (s *ScheduleTaskService) Create(ctx context.Context, req ScheduleTaskRequest) (*ScheduleTaskResponse, error) {
@@ -274,6 +186,7 @@ func (s *ScheduleTaskService) Create(ctx context.Context, req ScheduleTaskReques
Result: &result,
})
invalidateScheduleTaskCaches(ctx, s.cache, actor.TenantID, &id)
return &ScheduleTaskResponse{
ID: id, PromptRuleID: req.PromptRuleID, BrandID: req.BrandID,
Name: req.Name, CronExpr: req.CronExpr, TargetPlatform: req.TargetPlatform,
@@ -335,6 +248,7 @@ func (s *ScheduleTaskService) Update(ctx context.Context, id int64, req Schedule
if err := tx.Commit(ctx); err != nil {
return response.ErrInternal(50010, "update_failed", "failed to commit schedule task")
}
invalidateScheduleTaskCaches(ctx, s.cache, actor.TenantID, &id)
return nil
}
@@ -363,6 +277,7 @@ func (s *ScheduleTaskService) Delete(ctx context.Context, id int64) error {
AfterJSON: afterJSON,
Result: &result,
})
invalidateScheduleTaskCaches(ctx, s.cache, actor.TenantID, &id)
return nil
}
@@ -405,9 +320,123 @@ func (s *ScheduleTaskService) ToggleStatus(ctx context.Context, id int64, req Sc
if err := tx.Commit(ctx); err != nil {
return response.ErrInternal(50010, "update_failed", "failed to commit schedule task")
}
invalidateScheduleTaskCaches(ctx, s.cache, actor.TenantID, &id)
return nil
}
func (s *ScheduleTaskService) loadScheduleTasks(ctx context.Context, tenantID int64, params ScheduleTaskListParams) (*ScheduleTaskListResponse, error) {
if params.Page < 1 {
params.Page = 1
}
if params.PageSize < 1 || params.PageSize > 100 {
params.PageSize = 20
}
offset := (params.Page - 1) * params.PageSize
var total int64
if err := s.pool.QueryRow(ctx, `
SELECT COUNT(*) FROM schedule_tasks
WHERE tenant_id = $1 AND deleted_at IS NULL
AND ($2::bigint IS NULL OR prompt_rule_id = $2)
AND ($3::text IS NULL OR status = $3)
AND ($4::text IS NULL OR name ILIKE '%' || $4 || '%')
AND ($5::timestamptz IS NULL OR created_at >= $5)
AND ($6::timestamptz IS NULL OR created_at <= $6)
`, tenantID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo).Scan(&total); err != nil {
return nil, response.ErrInternal(50010, "count_failed", "failed to count schedule tasks")
}
rows, err := s.pool.Query(ctx, `
SELECT st.id, st.prompt_rule_id, st.brand_id, st.name,
st.cron_expr, st.target_platform, st.enable_web_search, st.generate_count, st.start_at, st.end_at,
st.next_run_at, st.status, st.created_at, st.updated_at,
pr.name AS prompt_rule_name,
b.name AS brand_name
FROM schedule_tasks st
LEFT JOIN prompt_rules pr ON pr.id = st.prompt_rule_id
LEFT JOIN brands b ON b.id = st.brand_id
WHERE st.tenant_id = $1 AND st.deleted_at IS NULL
AND ($2::bigint IS NULL OR st.prompt_rule_id = $2)
AND ($3::text IS NULL OR st.status = $3)
AND ($4::text IS NULL OR st.name ILIKE '%' || $4 || '%')
AND ($5::timestamptz IS NULL OR st.created_at >= $5)
AND ($6::timestamptz IS NULL OR st.created_at <= $6)
ORDER BY st.created_at DESC
LIMIT $7 OFFSET $8
`, tenantID, params.PromptRuleID, params.Status, params.Keyword, params.CreatedFrom, params.CreatedTo, params.PageSize, offset)
if err != nil {
return nil, response.ErrInternal(50010, "query_failed", "failed to list schedule tasks")
}
defer rows.Close()
items := make([]ScheduleTaskResponse, 0)
for rows.Next() {
item, err := scanScheduleTaskResponse(rows)
if err != nil {
return nil, err
}
items = append(items, item)
}
return &ScheduleTaskListResponse{Items: items, Total: total}, nil
}
func (s *ScheduleTaskService) loadScheduleTaskDetail(ctx context.Context, tenantID, taskID int64) (*ScheduleTaskResponse, bool, error) {
row := s.pool.QueryRow(ctx, `
SELECT st.id, st.prompt_rule_id, st.brand_id, st.name,
st.cron_expr, st.target_platform, st.enable_web_search, st.generate_count, st.start_at, st.end_at,
st.next_run_at, st.status, st.created_at, st.updated_at,
pr.name AS prompt_rule_name,
b.name AS brand_name
FROM schedule_tasks st
LEFT JOIN prompt_rules pr ON pr.id = st.prompt_rule_id
LEFT JOIN brands b ON b.id = st.brand_id
WHERE st.id = $1 AND st.tenant_id = $2 AND st.deleted_at IS NULL
`, taskID, tenantID)
item, err := scanScheduleTaskResponse(row)
if err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return nil, false, nil
}
return nil, false, err
}
return &item, true, nil
}
func scanScheduleTaskResponse(scanner interface {
Scan(dest ...interface{}) error
}) (ScheduleTaskResponse, error) {
var item ScheduleTaskResponse
var createdAt interface{}
var updatedAt interface{}
var startAt interface{}
var endAt interface{}
var nextRunAt interface{}
if err := scanner.Scan(&item.ID, &item.PromptRuleID, &item.BrandID, &item.Name,
&item.CronExpr, &item.TargetPlatform, &item.EnableWebSearch, &item.GenerateCount, &startAt, &endAt,
&nextRunAt, &item.Status, &createdAt, &updatedAt,
&item.PromptRuleName, &item.BrandName); err != nil {
if errors.Is(err, pgx.ErrNoRows) {
return ScheduleTaskResponse{}, pgx.ErrNoRows
}
return ScheduleTaskResponse{}, response.ErrInternal(50010, "scan_failed", err.Error())
}
item.CreatedAt = fmt.Sprintf("%v", createdAt)
item.UpdatedAt = fmt.Sprintf("%v", updatedAt)
if startAt != nil {
value := fmt.Sprintf("%v", startAt)
item.StartAt = &value
}
if endAt != nil {
value := fmt.Sprintf("%v", endAt)
item.EndAt = &value
}
if nextRunAt != nil {
value := fmt.Sprintf("%v", nextRunAt)
item.NextRunAt = &value
}
return item, nil
}
func (s *ScheduleTaskService) ensurePromptRuleExists(ctx context.Context, tenantID, promptRuleID int64) error {
var exists bool
_ = s.pool.QueryRow(ctx, `