数据存储

PostgreSQL 16 是事实源 · Valkey 8 只做缓存/限流/短锁 · COS 存文件和备份

---

存储全景

api-server / admin-api / worker
       │
       ├── PostgreSQL 16
       │   ├── users / products / inventory / orders / payments
       │   ├── outbox_events
       │   ├── idempotency_keys
       │   └── audit_logs
       │
       ├── Valkey 8
       │   ├── token version / blacklist
       │   ├── rate limit
       │   ├── short locks
       │   └── hot cache
       │
       └── COS + CDN
           ├── 商品图片
           ├── 检疫证明
           ├── 评价图片
           └── PostgreSQL 备份

原则：Valkey 丢失不能导致订单、支付、库存事实丢失。所有不可丢数据必须落 PostgreSQL。

---

数据 Ownership

模块	Owner 表
auth	user_credentials, admin_accounts, admin_roles, admin_permissions
user	users, user_addresses, points_logs
catalog	products, categories, trace_records
inventory	product_batches, inventory, inventory_locks, inventory_movements
order	orders, order_items, after_sales, reviews
payment	payments, payment_notifications, refunds
coupon	coupon_templates, user_coupons
delivery	delivery_tasks
asset	assets
shared	outbox_events, outbox_consumed, idempotency_keys, audit_logs

非 owner 模块不能直接写其他模块的表。跨模块写入必须调用 owner service。

---

用户与认证

CREATE TABLE users (
    id             BIGSERIAL PRIMARY KEY,
    nickname       VARCHAR(64),
    avatar_url     VARCHAR(512),
    phone          VARCHAR(256),              -- 加密值
    phone_hash     VARCHAR(128),              -- 查询用 hash
    member_level   VARCHAR(32) NOT NULL DEFAULT 'free',
    points         INT NOT NULL DEFAULT 0,
    status         VARCHAR(32) NOT NULL DEFAULT 'active',
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE UNIQUE INDEX idx_users_phone_hash ON users(phone_hash) WHERE phone_hash IS NOT NULL;

CREATE TABLE user_credentials (
    id              BIGSERIAL PRIMARY KEY,
    user_id         BIGINT NOT NULL REFERENCES users(id),
    auth_type       VARCHAR(32) NOT NULL,       -- wechat / password / driver
    openid          VARCHAR(128),
    unionid         VARCHAR(128),
    password_hash   VARCHAR(256),
    token_version   BIGINT NOT NULL DEFAULT 0,
    status          VARCHAR(32) NOT NULL DEFAULT 'active',
    locked_until    TIMESTAMPTZ,
    last_login_at   TIMESTAMPTZ,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE UNIQUE INDEX idx_creds_openid ON user_credentials(openid) WHERE openid IS NOT NULL;
CREATE INDEX idx_creds_user ON user_credentials(user_id);

CREATE TABLE user_addresses (
    id             BIGSERIAL PRIMARY KEY,
    user_id        BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    tag            VARCHAR(32),
    contact        VARCHAR(64) NOT NULL,
    phone          VARCHAR(256) NOT NULL,       -- 加密值
    phone_hash     VARCHAR(128) NOT NULL,
    province       VARCHAR(32),
    city           VARCHAR(32),
    district       VARCHAR(32),
    detail         TEXT NOT NULL,
    lat            DOUBLE PRECISION,
    lng            DOUBLE PRECISION,
    is_default     BOOLEAN NOT NULL DEFAULT false,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_addresses_user ON user_addresses(user_id);

---

商品与溯源

CREATE TABLE categories (
    id          BIGSERIAL PRIMARY KEY,
    name        VARCHAR(64) NOT NULL,
    code        VARCHAR(64) UNIQUE NOT NULL,
    sort_order  INT NOT NULL DEFAULT 0,
    status      VARCHAR(32) NOT NULL DEFAULT 'active'
);

CREATE TABLE products (
    id             BIGSERIAL PRIMARY KEY,
    category_id    BIGINT REFERENCES categories(id),
    name           VARCHAR(128) NOT NULL,
    part           VARCHAR(64),
    price_fen      BIGINT NOT NULL,
    unit           VARCHAR(32) NOT NULL DEFAULT '500g',
    description    TEXT,
    images         JSONB NOT NULL DEFAULT '[]',
    status         VARCHAR(32) NOT NULL DEFAULT 'draft',  -- draft / active / off_shelf
    sales          INT NOT NULL DEFAULT 0,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_products_category_status ON products(category_id, status);
CREATE INDEX idx_products_search ON products USING GIN(to_tsvector('simple', name || ' ' || coalesce(description, '')));

CREATE TABLE trace_records (
    id                 BIGSERIAL PRIMARY KEY,
    product_id         BIGINT NOT NULL REFERENCES products(id),
    batch_no           VARCHAR(64) NOT NULL,
    farm_name          VARCHAR(128) NOT NULL,
    farm_address       TEXT,
    breed              VARCHAR(64),
    feed_info          TEXT,
    slaughter_date     DATE NOT NULL,
    quarantine_cert_asset_id BIGINT,
    created_at         TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_trace_product ON trace_records(product_id);
CREATE INDEX idx_trace_batch ON trace_records(batch_no);

---

库存

生鲜库存必须按批次管理，并且每一次变动必须有流水。

CREATE TABLE product_batches (
    id                 BIGSERIAL PRIMARY KEY,
    product_id          BIGINT NOT NULL REFERENCES products(id),
    batch_no            VARCHAR(64) UNIQUE NOT NULL,
    supplier            VARCHAR(128),
    slaughter_date      DATE,
    production_date     DATE,
    expire_at           TIMESTAMPTZ,
    status              VARCHAR(32) NOT NULL DEFAULT 'active',
    created_at          TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_batches_product ON product_batches(product_id, status);

CREATE TABLE inventory (
    id              BIGSERIAL PRIMARY KEY,
    product_id      BIGINT NOT NULL REFERENCES products(id),
    batch_no        VARCHAR(64) NOT NULL REFERENCES product_batches(batch_no),
    total_qty       INT NOT NULL DEFAULT 0,
    reserved_qty    INT NOT NULL DEFAULT 0,
    version         BIGINT NOT NULL DEFAULT 0,
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    UNIQUE(product_id, batch_no)
);
CREATE INDEX idx_inventory_product ON inventory(product_id);

CREATE TABLE inventory_locks (
    lock_id       VARCHAR(64) PRIMARY KEY,
    order_no      VARCHAR(32) NOT NULL,
    product_id    BIGINT NOT NULL,
    batch_no      VARCHAR(64) NOT NULL,
    quantity      INT NOT NULL,
    status        VARCHAR(32) NOT NULL DEFAULT 'active', -- active / confirmed / released
    created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    confirmed_at  TIMESTAMPTZ,
    released_at   TIMESTAMPTZ
);
CREATE INDEX idx_inventory_locks_order ON inventory_locks(order_no);

CREATE TABLE inventory_movements (
    id                 BIGSERIAL PRIMARY KEY,
    product_id          BIGINT NOT NULL,
    batch_no            VARCHAR(64) NOT NULL,
    movement_type       VARCHAR(32) NOT NULL, -- inbound / lock / confirm / release / refund_rollback / adjust
    quantity            INT NOT NULL,
    lock_id             VARCHAR(64),
    order_no            VARCHAR(32),
    operator_id         BIGINT DEFAULT 0,
    balance_total       INT NOT NULL,
    balance_reserved    INT NOT NULL,
    note                TEXT,
    created_at          TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_movements_product ON inventory_movements(product_id, created_at);
CREATE INDEX idx_movements_order ON inventory_movements(order_no);
CREATE INDEX idx_movements_lock ON inventory_movements(lock_id);

---

订单与售后

CREATE TABLE orders (
    id                 BIGSERIAL PRIMARY KEY,
    order_no           VARCHAR(32) UNIQUE NOT NULL,
    user_id            BIGINT NOT NULL REFERENCES users(id),
    address_snapshot   JSONB NOT NULL,
    total_fen          BIGINT NOT NULL,
    discount_fen       BIGINT NOT NULL DEFAULT 0,
    delivery_fee_fen   BIGINT NOT NULL DEFAULT 0,
    final_fen          BIGINT NOT NULL,
    status             VARCHAR(32) NOT NULL DEFAULT 'created',
    stock_lock_id      VARCHAR(64),
    coupon_id          BIGINT,
    payment_no         VARCHAR(64),
    delivery_time      TSRANGE,
    remark             TEXT,
    version            BIGINT NOT NULL DEFAULT 0,
    paid_at            TIMESTAMPTZ,
    cancelled_at       TIMESTAMPTZ,
    delivered_at       TIMESTAMPTZ,
    completed_at       TIMESTAMPTZ,
    created_at         TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at         TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_orders_user_created ON orders(user_id, created_at DESC);
CREATE INDEX idx_orders_status_created ON orders(status, created_at);

CREATE TABLE order_items (
    id              BIGSERIAL PRIMARY KEY,
    order_id        BIGINT NOT NULL REFERENCES orders(id) ON DELETE CASCADE,
    product_id      BIGINT NOT NULL REFERENCES products(id),
    product_name    VARCHAR(128) NOT NULL,
    price_fen       BIGINT NOT NULL,
    quantity        INT NOT NULL,
    batch_no        VARCHAR(64),
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_order_items_order ON order_items(order_id);

CREATE TABLE after_sales (
    id             BIGSERIAL PRIMARY KEY,
    order_no       VARCHAR(32) NOT NULL,
    user_id        BIGINT NOT NULL,
    type           VARCHAR(32) NOT NULL, -- refund / return / complaint
    reason         TEXT NOT NULL,
    images         JSONB NOT NULL DEFAULT '[]',
    status         VARCHAR(32) NOT NULL DEFAULT 'applied',
    refund_fen     BIGINT NOT NULL DEFAULT 0,
    admin_remark   TEXT,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_after_sales_order ON after_sales(order_no);

address_snapshot 是下单时的地址快照，不能依赖用户后续修改后的地址。

---

支付与退款

CREATE TABLE payments (
    id                BIGSERIAL PRIMARY KEY,
    payment_no        VARCHAR(64) UNIQUE NOT NULL,
    order_no          VARCHAR(32) UNIQUE NOT NULL,
    user_id           BIGINT NOT NULL,
    amount_fen        BIGINT NOT NULL,
    channel           VARCHAR(32) NOT NULL DEFAULT 'wechat',
    status            VARCHAR(32) NOT NULL DEFAULT 'created',
    wechat_prepay_id  VARCHAR(128),
    wechat_txn_id     VARCHAR(128),
    paid_at           TIMESTAMPTZ,
    version           BIGINT NOT NULL DEFAULT 0,
    created_at        TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at        TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE UNIQUE INDEX idx_payments_wechat_txn ON payments(wechat_txn_id) WHERE wechat_txn_id IS NOT NULL;
CREATE INDEX idx_payments_status ON payments(status, created_at);

CREATE TABLE payment_notifications (
    id             BIGSERIAL PRIMARY KEY,
    notify_id      VARCHAR(128) UNIQUE NOT NULL,
    payment_no     VARCHAR(64),
    order_no       VARCHAR(32),
    channel        VARCHAR(32) NOT NULL,
    event_type     VARCHAR(64) NOT NULL,
    raw_headers    JSONB NOT NULL,
    raw_body       JSONB NOT NULL,
    verified       BOOLEAN NOT NULL DEFAULT false,
    processed      BOOLEAN NOT NULL DEFAULT false,
    error_msg      TEXT,
    created_at     TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_payment_notifications_order ON payment_notifications(order_no);

CREATE TABLE refunds (
    id              BIGSERIAL PRIMARY KEY,
    refund_no       VARCHAR(64) UNIQUE NOT NULL,
    payment_no      VARCHAR(64) NOT NULL,
    order_no        VARCHAR(32) NOT NULL,
    amount_fen      BIGINT NOT NULL,
    reason          TEXT NOT NULL,
    status          VARCHAR(32) NOT NULL DEFAULT 'created',
    wechat_refund_id VARCHAR(128),
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_refunds_order ON refunds(order_no);

支付回调原始数据必须留存，便于对账和事故复盘。

---

优惠券与配送

CREATE TABLE coupon_templates (
    id              BIGSERIAL PRIMARY KEY,
    name            VARCHAR(64) NOT NULL,
    type            VARCHAR(32) NOT NULL, -- fixed / discount / free_shipping
    value           INT NOT NULL,
    min_amount_fen  BIGINT NOT NULL DEFAULT 0,
    total_qty       INT NOT NULL,
    remain_qty      INT NOT NULL DEFAULT 0,
    start_at        TIMESTAMPTZ NOT NULL,
    end_at          TIMESTAMPTZ NOT NULL,
    status          VARCHAR(32) NOT NULL DEFAULT 'active',
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE TABLE user_coupons (
    id            BIGSERIAL PRIMARY KEY,
    user_id       BIGINT NOT NULL REFERENCES users(id),
    template_id   BIGINT NOT NULL REFERENCES coupon_templates(id),
    status        VARCHAR(32) NOT NULL DEFAULT 'unused', -- unused / locked / used / expired
    locked_at     TIMESTAMPTZ,
    order_no      VARCHAR(32),
    used_at       TIMESTAMPTZ,
    expires_at    TIMESTAMPTZ NOT NULL,
    created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_user_coupons_user_status ON user_coupons(user_id, status);

CREATE TABLE delivery_tasks (
    id                BIGSERIAL PRIMARY KEY,
    order_no          VARCHAR(32) UNIQUE NOT NULL,
    time_slot         VARCHAR(32) NOT NULL,
    status            VARCHAR(32) NOT NULL DEFAULT 'pending', -- pending / claimed / picked_up / delivered / cancelled
    driver_id         BIGINT,
    address_summary   VARCHAR(256) NOT NULL,
    contact_masked    VARCHAR(64) NOT NULL,
    phone_masked      VARCHAR(32) NOT NULL,
    lat               DOUBLE PRECISION,
    lng               DOUBLE PRECISION,
    remark            TEXT,
    claimed_at        TIMESTAMPTZ,
    picked_up_at      TIMESTAMPTZ,
    delivered_at      TIMESTAMPTZ,
    created_at        TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at        TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_delivery_slot_status ON delivery_tasks(time_slot, status);
CREATE INDEX idx_delivery_driver_status ON delivery_tasks(driver_id, status);

---

Outbox、幂等、审计

CREATE TABLE outbox_events (
    id              BIGSERIAL PRIMARY KEY,
    event_id        UUID UNIQUE NOT NULL,
    event_type      VARCHAR(64) NOT NULL,
    aggregate_type  VARCHAR(64) NOT NULL,
    aggregate_id    VARCHAR(128) NOT NULL,
    payload         JSONB NOT NULL DEFAULT '{}',
    status          VARCHAR(32) NOT NULL DEFAULT 'pending',
    retry_count     INT NOT NULL DEFAULT 0,
    max_retries     INT NOT NULL DEFAULT 10,
    scheduled_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    next_retry_at   TIMESTAMPTZ,
    locked_by       VARCHAR(64),
    locked_at       TIMESTAMPTZ,
    error_msg       TEXT,
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    processed_at    TIMESTAMPTZ
);
CREATE INDEX idx_outbox_pending ON outbox_events(status, scheduled_at, id);
CREATE INDEX idx_outbox_retry ON outbox_events(status, next_retry_at, id);

CREATE TABLE outbox_consumed (
    event_id      UUID NOT NULL,
    consumer      VARCHAR(64) NOT NULL,
    consumed_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    PRIMARY KEY (event_id, consumer)
);

CREATE TABLE idempotency_keys (
    key          VARCHAR(128) PRIMARY KEY,
    scope        VARCHAR(64) NOT NULL,
    owner_id     VARCHAR(64) NOT NULL,
    response     JSONB NOT NULL,
    created_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    expires_at   TIMESTAMPTZ NOT NULL
);
CREATE INDEX idx_idempotency_expires ON idempotency_keys(expires_at);

CREATE TABLE audit_logs (
    id              BIGSERIAL PRIMARY KEY,
    operator_id     BIGINT NOT NULL,
    operator_name   VARCHAR(64) NOT NULL,
    action          VARCHAR(64) NOT NULL,
    target_type     VARCHAR(64) NOT NULL,
    target_id       VARCHAR(128) NOT NULL,
    before_value    JSONB,
    after_value     JSONB,
    reason          TEXT NOT NULL,
    ip_address      VARCHAR(45),
    trace_id        VARCHAR(64),
    created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_audit_target ON audit_logs(target_type, target_id);
CREATE INDEX idx_audit_operator ON audit_logs(operator_id, created_at);

Worker 抢任务必须使用：

SELECT *
FROM outbox_events
WHERE status = 'pending'
  AND scheduled_at <= NOW()
ORDER BY scheduled_at, id
LIMIT 50
FOR UPDATE SKIP LOCKED;

---

Valkey 规则

Key	TTL	用途
token_version:{user_id}	无	登出/改密码后让旧 token 失效
token_blacklist:{jti}	15m	并发窗口内屏蔽旧 access token
rate:{scope}:{id}	1m	登录、下单、支付、退款限流
idempotency:{scope}:{owner}:{key}	24h	幂等结果热缓存
product:{id}	5m	商品详情缓存
product:hot	5m	热门商品列表
inventory:{product_id}:stock	30s	展示库存缓存，不作为事实
lock:timeout:{order_no}	30s	worker 超时处理短锁
lock:admin:{target}	30s	后台人工敏感操作短锁

---

COS 路径

类别	路径前缀	说明
商品图片	products/{product_id}/	商品主图和详情图
检疫证明	quarantine/{batch_no}/	图片或 PDF
用户头像	avatars/{user_id}/	头像
售后图片	after-sales/{order_no}/	用户上传凭证
数据库备份	backups/postgres/	pg_dump 归档

上传流程：前端请求预签名 URL → 直传 COS → COS 回调或前端确认 → asset 模块写 assets 表。

---

备份与恢复

对象	策略
PostgreSQL 全量	每日 pg_dump -Fc 到 COS，保留 30 天
PostgreSQL WAL	生产期启用 WAL 归档，保留 7 天
COS 文件	开启版本控制或定期清单
Valkey	RDB 快照即可，丢失可接受

目标：RPO < 24 小时，RTO < 4 小时。每月至少做一次恢复演练：新建空库，从备份恢复，跑核心冒烟测试。

---

相关链接

• 技术架构总览
• 模块架构
• 状态机设计
• API 设计