sql数据流，amdin业务逻辑接入

docs/sql/30_rpc/marketing/rpc_admin_get_integral_stats_v1.sql

@@ -0,0 +1,91 @@
+-- RPC: rpc_admin_get_integral_stats
+-- 位置：docs/sql/30_rpc/marketing/rpc_admin_get_integral_stats_v1.sql
+-- 说明：聚合统计积分概况（总额、趋势、分布）
+
+CREATE OR REPLACE FUNCTION public.rpc_admin_get_integral_stats(
+    p_start_time TIMESTAMP WITH TIME ZONE,
+    p_end_time TIMESTAMP WITH TIME ZONE
+)
+RETURNS JSONB
+SECURITY DEFINER
+SET search_path = public
+LANGUAGE plpgsql
+AS $$
+DECLARE
+    v_total_stats RECORD;
+    v_trend_data JSONB;
+    v_source_dist JSONB;
+    v_consume_dist JSONB;
+BEGIN
+    -- 1. 权限检查
+    IF NOT EXISTS (
+        SELECT 1 FROM public.ak_users 
+        WHERE id = auth.uid() AND role = 'admin'
+    ) THEN
+        RAISE EXCEPTION 'Permission denied';
+    END IF;
+
+    -- 2. 计算核心指标 (所有时间)
+    SELECT 
+        COALESCE(SUM(CASE WHEN pm = 1 THEN number ELSE -number END), 0) as current_total,
+        COALESCE(SUM(CASE WHEN pm = 1 THEN number ELSE 0 END), 0) as cumulative_income,
+        COALESCE(SUM(CASE WHEN pm = 0 THEN number ELSE 0 END), 0) as cumulative_expend
+    INTO v_total_stats
+    FROM public.ml_user_bill
+    WHERE category = 'integral' AND status = 1;
+
+    -- 3. 趋势数据 (按日聚合)
+    SELECT jsonb_agg(t) INTO v_trend_data
+    FROM (
+        SELECT
+            to_char(date_trunc('day', gs.day), 'MM-DD') AS date_group,
+            COALESCE((SELECT SUM(number) FROM public.ml_user_bill b 
+                      WHERE b.category = 'integral' AND b.pm = 1 AND b.status = 1 
+                      AND date_trunc('day', b.created_at) = gs.day), 0) as income,
+            COALESCE((SELECT SUM(number) FROM public.ml_user_bill b 
+                      WHERE b.category = 'integral' AND b.pm = 0 AND b.status = 1 
+                      AND date_trunc('day', b.created_at) = gs.day), 0) as expend
+        FROM generate_series(date_trunc('day', p_start_time), date_trunc('day', p_end_time), '1 day'::interval) gs(day)
+        ORDER BY gs.day ASC
+    ) t;
+
+    -- 4. 来源分布 (按 type 分组)
+    SELECT jsonb_agg(t) INTO v_source_dist
+    FROM (
+        SELECT 
+            type as label, 
+            SUM(number) as value,
+            ROUND((SUM(number) * 100 / NULLIF(v_total_stats.cumulative_income, 0)), 2) as percent
+        FROM public.ml_user_bill
+        WHERE category = 'integral' AND pm = 1 AND status = 1
+        GROUP BY type
+        ORDER BY value DESC
+    ) t;
+
+    -- 5. 消耗分布 (按 type 分组)
+    SELECT jsonb_agg(t) INTO v_consume_dist
+    FROM (
+        SELECT 
+            type as label, 
+            SUM(number) as value,
+            ROUND((SUM(number) * 100 / NULLIF(v_total_stats.cumulative_expend, 0)), 2) as percent
+        FROM public.ml_user_bill
+        WHERE category = 'integral' AND pm = 0 AND status = 1
+        GROUP BY type
+        ORDER BY value DESC
+    ) t;
+
+    RETURN jsonb_build_object(
+        'totals', jsonb_build_object(
+            'current', v_total_stats.current_total,
+            'income', v_total_stats.cumulative_income,
+            'expend', v_total_stats.cumulative_expend
+        ),
+        'trend', COALESCE(v_trend_data, '[]'::jsonb),
+        'sources', COALESCE(v_source_dist, '[]'::jsonb),
+        'consumes', COALESCE(v_consume_dist, '[]'::jsonb)
+    );
+END;
+$$;
+
+GRANT EXECUTE ON FUNCTION public.rpc_admin_get_integral_stats(timestamptz, timestamptz) TO authenticated;