# Contribution Service (贡献值/算力服务) 开发指导 ## 1. 服务概述 ### 1.1 核心职责 Contribution Service 负责管理用户的贡献值(算力),这是挖矿系统的核心计算基础。 **主要功能:** - 通过 Debezium CDC 同步用户、认种、推荐关系数据 - 计算用户算力(来自自己认种 + 团队贡献) - 维护算力明细账(每笔算力的来源可追溯) - 处理算力过期(2年有效期) - 管理未分配算力归总部逻辑 ### 1.2 技术栈 - **框架**: NestJS + TypeScript - **数据库**: PostgreSQL (事务型) - **ORM**: Prisma - **消息队列**: Kafka (Debezium CDC + 事件发布) - **缓存**: Redis ### 1.3 端口分配 - HTTP: 3020 - 数据库: rwa_contribution --- ## 2. 架构设计 ### 2.1 六边形架构分层 ``` ┌─────────────────────────────────────────────────────────────┐ │ API Layer (api/) │ │ Controllers, DTOs - 处理 HTTP 请求 │ ├─────────────────────────────────────────────────────────────┤ │ Application Layer (application/) │ │ Commands, Queries, Event Handlers - 业务流程编排 │ ├─────────────────────────────────────────────────────────────┤ │ Domain Layer (domain/) │ │ Aggregates, Value Objects, Domain Events - 核心业务规则 │ ├─────────────────────────────────────────────────────────────┤ │ Infrastructure Layer (infrastructure/) │ │ Prisma, Kafka, Redis - 技术实现细节 │ └─────────────────────────────────────────────────────────────┘ ``` ### 2.2 目录结构 ``` contribution-service/ ├── src/ │ ├── api/ # API层 │ │ ├── controllers/ │ │ │ ├── contribution.controller.ts # 算力查询API │ │ │ ├── sync-status.controller.ts # 同步状态API │ │ │ └── health.controller.ts │ │ └── dto/ │ │ ├── request/ │ │ └── response/ │ │ ├── contribution-account.response.ts │ │ └── contribution-detail.response.ts │ │ │ ├── application/ # 应用层 │ │ ├── commands/ │ │ │ ├── calculate-user-contribution.command.ts │ │ │ ├── process-adoption-contribution.command.ts │ │ │ ├── expire-contributions.command.ts │ │ │ └── recalculate-all-contributions.command.ts │ │ ├── queries/ │ │ │ ├── get-user-contribution.query.ts │ │ │ ├── get-contribution-details.query.ts │ │ │ └── get-network-total-contribution.query.ts │ │ ├── services/ │ │ │ └── contribution-calculation.service.ts │ │ ├── event-handlers/ │ │ │ ├── adoption-synced.handler.ts │ │ │ ├── user-synced.handler.ts │ │ │ └── referral-synced.handler.ts │ │ └── schedulers/ │ │ ├── contribution-expiry.scheduler.ts │ │ └── daily-snapshot.scheduler.ts │ │ │ ├── domain/ # 领域层 │ │ ├── aggregates/ │ │ │ ├── contribution-account.aggregate.ts │ │ │ └── contribution-record.aggregate.ts │ │ ├── repositories/ │ │ │ ├── contribution-account.repository.interface.ts │ │ │ ├── contribution-record.repository.interface.ts │ │ │ ├── synced-user.repository.interface.ts │ │ │ ├── synced-adoption.repository.interface.ts │ │ │ └── synced-referral.repository.interface.ts │ │ ├── value-objects/ │ │ │ ├── contribution-amount.vo.ts │ │ │ ├── distribution-rate.vo.ts │ │ │ └── account-sequence.vo.ts │ │ ├── events/ │ │ │ ├── contribution-calculated.event.ts │ │ │ ├── contribution-expired.event.ts │ │ │ └── daily-snapshot-created.event.ts │ │ └── services/ │ │ ├── contribution-calculator.service.ts │ │ └── team-contribution-calculator.service.ts │ │ │ ├── infrastructure/ # 基础设施层 │ │ ├── persistence/ │ │ │ ├── prisma/ │ │ │ │ └── prisma.service.ts │ │ │ ├── repositories/ │ │ │ │ ├── contribution-account.repository.impl.ts │ │ │ │ ├── contribution-record.repository.impl.ts │ │ │ │ ├── synced-user.repository.impl.ts │ │ │ │ ├── synced-adoption.repository.impl.ts │ │ │ │ └── synced-referral.repository.impl.ts │ │ │ └── unit-of-work/ │ │ │ └── unit-of-work.service.ts │ │ ├── kafka/ │ │ │ ├── cdc-consumers/ │ │ │ │ ├── user-cdc.consumer.ts │ │ │ │ ├── adoption-cdc.consumer.ts │ │ │ │ └── referral-cdc.consumer.ts │ │ │ ├── event-publisher.service.ts │ │ │ └── kafka.module.ts │ │ ├── redis/ │ │ │ └── contribution-cache.service.ts │ │ └── infrastructure.module.ts │ │ │ ├── shared/ │ ├── config/ │ ├── app.module.ts │ └── main.ts │ ├── prisma/ │ ├── schema.prisma │ └── migrations/ ├── package.json ├── tsconfig.json ├── Dockerfile └── docker-compose.yml ``` --- ## 3. 数据库设计 ### 3.1 数据库类型选择 | 表类型 | 数据库类型 | 原因 | |--------|-----------|------| | 同步数据表 | 事务型 (PostgreSQL) | CDC 数据需要精确同步,支持事务 | | 算力账户表 | 事务型 (PostgreSQL) | 余额变更需要强一致性 | | 算力明细表 | 事务型 (PostgreSQL) | 明细账需要完整性约束 | | 快照表 | 事务型 (PostgreSQL) | 历史数据需要持久化 | ### 3.2 核心表结构 ```sql -- ============================================ -- CDC 同步数据表(从其他服务同步) -- ============================================ -- 同步的用户数据 CREATE TABLE synced_users ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), account_sequence VARCHAR(20) NOT NULL UNIQUE, -- 跨服务关联键 original_user_id UUID NOT NULL, phone VARCHAR(20), status VARCHAR(20), created_at TIMESTAMP WITH TIME ZONE, -- CDC 同步元数据 source_sequence_num BIGINT NOT NULL, -- 源数据的序列号 synced_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), -- 算力计算状态 contribution_calculated BOOLEAN DEFAULT FALSE, contribution_calculated_at TIMESTAMP WITH TIME ZONE ); CREATE INDEX idx_synced_users_sequence ON synced_users(account_sequence); CREATE INDEX idx_synced_users_not_calculated ON synced_users(contribution_calculated) WHERE contribution_calculated = FALSE; -- 同步的认种数据 CREATE TABLE synced_adoptions ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), original_adoption_id UUID NOT NULL UNIQUE, account_sequence VARCHAR(20) NOT NULL, tree_count INT NOT NULL, adoption_date DATE NOT NULL, status VARCHAR(20), -- 贡献值计算参数(从认种时的配置) contribution_per_tree DECIMAL(20,10) NOT NULL, -- CDC 同步元数据 source_sequence_num BIGINT NOT NULL, synced_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), -- 算力分配状态 contribution_distributed BOOLEAN DEFAULT FALSE, contribution_distributed_at TIMESTAMP WITH TIME ZONE ); CREATE INDEX idx_synced_adoptions_account ON synced_adoptions(account_sequence); CREATE INDEX idx_synced_adoptions_not_distributed ON synced_adoptions(contribution_distributed) WHERE contribution_distributed = FALSE; -- 同步的推荐关系数据 CREATE TABLE synced_referrals ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), account_sequence VARCHAR(20) NOT NULL, -- 用户 referrer_account_sequence VARCHAR(20), -- 推荐人 -- 预计算的层级路径(便于快速查询上下级) ancestor_path TEXT, -- 格式: /root/seq1/seq2/.../ depth INT DEFAULT 0, -- CDC 同步元数据 source_sequence_num BIGINT NOT NULL, synced_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), UNIQUE(account_sequence) ); CREATE INDEX idx_synced_referrals_referrer ON synced_referrals(referrer_account_sequence); CREATE INDEX idx_synced_referrals_path ON synced_referrals USING gin(ancestor_path gin_trgm_ops); -- ============================================ -- 算力账户与明细表 -- ============================================ -- 算力账户表(汇总) CREATE TABLE contribution_accounts ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), account_sequence VARCHAR(20) NOT NULL UNIQUE, -- 算力汇总 personal_contribution DECIMAL(30,10) DEFAULT 0, -- 来自自己认种 (70%) team_level_contribution DECIMAL(30,10) DEFAULT 0, -- 来自团队层级 (0.5%×N级) team_bonus_contribution DECIMAL(30,10) DEFAULT 0, -- 来自团队额外奖励 (2.5%×N) total_contribution DECIMAL(30,10) DEFAULT 0, -- 总算力 effective_contribution DECIMAL(30,10) DEFAULT 0, -- 有效算力(未过期) -- 用户条件(决定能获得多少团队算力) has_adopted BOOLEAN DEFAULT FALSE, direct_referral_adopted_count INT DEFAULT 0, -- 解锁状态 unlocked_level_depth INT DEFAULT 0, -- 5/10/15 unlocked_bonus_tiers INT DEFAULT 0, -- 1/2/3 -- 版本号(乐观锁) version INT DEFAULT 1, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); -- 算力明细表(分类账) CREATE TABLE contribution_records ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), account_sequence VARCHAR(20) NOT NULL, -- 算力归属用户 -- 来源信息(可追溯) source_type VARCHAR(30) NOT NULL, -- PERSONAL / TEAM_LEVEL / TEAM_BONUS source_adoption_id UUID NOT NULL, -- 来源认种记录 source_account_sequence VARCHAR(20) NOT NULL, -- 认种人 -- 计算参数(审计用) tree_count INT NOT NULL, base_contribution DECIMAL(20,10) NOT NULL, distribution_rate DECIMAL(10,6) NOT NULL, -- 70% / 0.5% / 2.5% level_depth INT, -- 层级(TEAM_LEVEL时) bonus_tier INT, -- 档位(TEAM_BONUS时,1/2/3) -- 结果 amount DECIMAL(30,10) NOT NULL, -- 有效期 effective_date DATE NOT NULL, -- 次日生效 expire_date DATE NOT NULL, -- 2年后过期 is_expired BOOLEAN DEFAULT FALSE, expired_at TIMESTAMP WITH TIME ZONE, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); CREATE INDEX idx_contribution_records_account ON contribution_records(account_sequence); CREATE INDEX idx_contribution_records_source ON contribution_records(source_adoption_id); CREATE INDEX idx_contribution_records_expire ON contribution_records(expire_date) WHERE is_expired = FALSE; -- 未分配算力记录(归总部) CREATE TABLE unallocated_contributions ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), source_adoption_id UUID NOT NULL, source_account_sequence VARCHAR(20) NOT NULL, unalloc_type VARCHAR(30) NOT NULL, -- LEVEL_OVERFLOW / BONUS_TIER_1/2/3 would_be_account_sequence VARCHAR(20), -- 本应获得的上线 level_depth INT, amount DECIMAL(30,10) NOT NULL, reason VARCHAR(200), -- 归总部后的处理 allocated_to_headquarters BOOLEAN DEFAULT FALSE, allocated_at TIMESTAMP WITH TIME ZONE, effective_date DATE NOT NULL, expire_date DATE NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); -- 系统账户(运营/省/市/总部) CREATE TABLE system_accounts ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), account_type VARCHAR(20) NOT NULL UNIQUE, -- OPERATION/PROVINCE/CITY/HEADQUARTERS name VARCHAR(100) NOT NULL, contribution_balance DECIMAL(30,10) DEFAULT 0, contribution_never_expires BOOLEAN DEFAULT FALSE, version INT DEFAULT 1, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); -- 系统账户算力明细 CREATE TABLE system_contribution_records ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), system_account_id UUID NOT NULL REFERENCES system_accounts(id), source_adoption_id UUID NOT NULL, source_account_sequence VARCHAR(20) NOT NULL, distribution_rate DECIMAL(10,6) NOT NULL, -- 12% / 1% / 2% amount DECIMAL(30,10) NOT NULL, effective_date DATE NOT NULL, expire_date DATE, -- NULL = 永不过期 is_expired BOOLEAN DEFAULT FALSE, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); -- ============================================ -- 快照与统计表 -- ============================================ -- 每日算力快照(用于挖矿分配计算) CREATE TABLE daily_contribution_snapshots ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), snapshot_date DATE NOT NULL, account_sequence VARCHAR(20) NOT NULL, effective_contribution DECIMAL(30,10) NOT NULL, network_total_contribution DECIMAL(30,10) NOT NULL, contribution_ratio DECIMAL(30,18) NOT NULL, -- 占比(高精度) created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), UNIQUE(snapshot_date, account_sequence) ); CREATE INDEX idx_daily_snapshots_date ON daily_contribution_snapshots(snapshot_date); -- 用户团队统计(缓存,定期更新) CREATE TABLE user_team_stats ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), account_sequence VARCHAR(20) NOT NULL, stats_date DATE NOT NULL, -- 各级认种统计 level_1_trees INT DEFAULT 0, level_2_trees INT DEFAULT 0, level_3_trees INT DEFAULT 0, level_4_trees INT DEFAULT 0, level_5_trees INT DEFAULT 0, level_6_trees INT DEFAULT 0, level_7_trees INT DEFAULT 0, level_8_trees INT DEFAULT 0, level_9_trees INT DEFAULT 0, level_10_trees INT DEFAULT 0, level_11_trees INT DEFAULT 0, level_12_trees INT DEFAULT 0, level_13_trees INT DEFAULT 0, level_14_trees INT DEFAULT 0, level_15_trees INT DEFAULT 0, total_team_trees INT DEFAULT 0, direct_adopted_referrals INT DEFAULT 0, -- 直推认种用户数 created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(), UNIQUE(account_sequence, stats_date) ); -- ============================================ -- CDC 同步状态追踪 -- ============================================ -- CDC 同步进度表 CREATE TABLE cdc_sync_progress ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), source_topic VARCHAR(100) NOT NULL UNIQUE, last_sequence_num BIGINT DEFAULT 0, last_synced_at TIMESTAMP WITH TIME ZONE, updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); -- 已处理事件表(幂等性) CREATE TABLE processed_events ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), event_id VARCHAR(100) NOT NULL UNIQUE, event_type VARCHAR(50) NOT NULL, source_service VARCHAR(50), processed_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); -- ============================================ -- 配置表 -- ============================================ -- 贡献值递增配置 CREATE TABLE contribution_configs ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), base_contribution DECIMAL(20,10) DEFAULT 22617, increment_percentage DECIMAL(10,6) DEFAULT 0.003, -- 0.3% unit_size INT DEFAULT 100, start_tree_number INT DEFAULT 1000, is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); -- 分配比例配置 CREATE TABLE distribution_rate_configs ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), rate_type VARCHAR(30) NOT NULL UNIQUE, -- PERSONAL/OPERATION/PROVINCE/CITY/LEVEL_PER/BONUS_PER rate_value DECIMAL(10,6) NOT NULL, description VARCHAR(100), is_active BOOLEAN DEFAULT TRUE, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() ); ``` --- ## 4. 核心业务逻辑 ### 4.1 算力计算公式 ```typescript /** * 计算用户 X 的总算力 */ function calculateUserContribution(accountSequence: string): ContributionResult { const user = getUserWithTeamStats(accountSequence); // 1. 来自自己认种 (70%) const personalContribution = user.ownAdoptions.reduce((sum, adoption) => { return sum + adoption.treeCount * adoption.contributionPerTree * 0.70; }, 0); // 2. 来自团队层级 (每级0.5%) const unlockedLevels = getUnlockedLevelDepth(user.directReferralAdoptedCount); let teamLevelContribution = 0; for (let level = 1; level <= unlockedLevels; level++) { const levelTrees = user.teamStats[`level_${level}_trees`]; const levelContribution = levelTrees * baseContribution * 0.005; // 0.5% teamLevelContribution += levelContribution; } // 3. 来自团队额外奖励 (只看第1级,最多3个2.5%) const unlockedBonusTiers = getUnlockedBonusTiers(user); const level1Trees = user.teamStats.level_1_trees; const teamBonusContribution = level1Trees * baseContribution * 0.025 * unlockedBonusTiers; return { personalContribution, teamLevelContribution, teamBonusContribution, totalContribution: personalContribution + teamLevelContribution + teamBonusContribution }; } /** * 根据直推认种用户数确定解锁层级 */ function getUnlockedLevelDepth(directReferralAdoptedCount: number): number { if (directReferralAdoptedCount >= 5) return 15; if (directReferralAdoptedCount >= 3) return 10; if (directReferralAdoptedCount >= 1) return 5; return 0; } /** * 根据用户条件确定解锁的额外奖励档位数 */ function getUnlockedBonusTiers(user: User): number { let tiers = 0; if (user.hasAdopted) tiers++; // 自己认种过 → +1档 if (user.directReferralAdoptedCount >= 2) tiers++; // 直推≥2 → +1档 if (user.directReferralAdoptedCount >= 4) tiers++; // 直推≥4 → +1档 return tiers; // 0/1/2/3 } ``` ### 4.2 认种事件处理流程 ```typescript /** * 当新认种发生时的处理流程 */ async function processAdoptionContribution(adoption: SyncedAdoption): Promise { const totalContribution = adoption.treeCount * adoption.contributionPerTree; await unitOfWork.runInTransaction(async (tx) => { // 1. 分配给认种人 (70%) await createContributionRecord(tx, { accountSequence: adoption.accountSequence, sourceType: 'PERSONAL', sourceAdoptionId: adoption.id, amount: totalContribution * 0.70, distributionRate: 0.70, }); // 2. 分配给系统账户 (15%) await createSystemContribution(tx, 'OPERATION', adoption, 0.12); await createSystemContribution(tx, 'PROVINCE', adoption, 0.01); await createSystemContribution(tx, 'CITY', adoption, 0.02); // 3. 分配给上线团队 (15%) await distributeTeamContribution(tx, adoption, totalContribution * 0.15); // 4. 标记已分配 await markAdoptionDistributed(tx, adoption.id); }); } /** * 分配团队贡献值给上线链条 */ async function distributeTeamContribution( tx: Transaction, adoption: Adoption, teamTotal: number ): Promise { const ancestors = await getAncestorChain(adoption.accountSequence, 15); let distributedLevel = 0; let distributedBonus = 0; for (let i = 0; i < ancestors.length && i < 15; i++) { const ancestor = ancestors[i]; const level = i + 1; // 层级部分 (0.5% 每级) const levelAmount = teamTotal * 0.5 / 15; // 7.5% / 15 = 0.5% if (ancestor.unlockedLevelDepth >= level) { await createContributionRecord(tx, { accountSequence: ancestor.accountSequence, sourceType: 'TEAM_LEVEL', levelDepth: level, amount: levelAmount, }); distributedLevel += levelAmount; } else { // 未解锁,归总部 await createUnallocatedContribution(tx, { type: 'LEVEL_OVERFLOW', wouldBeAccount: ancestor.accountSequence, levelDepth: level, amount: levelAmount, }); } } // 额外奖励部分 (只给直接上线) if (ancestors.length > 0) { const directReferrer = ancestors[0]; const bonusPerTier = teamTotal * 0.5 / 3; // 7.5% / 3 = 2.5% for (let tier = 1; tier <= 3; tier++) { if (directReferrer.unlockedBonusTiers >= tier) { await createContributionRecord(tx, { accountSequence: directReferrer.accountSequence, sourceType: 'TEAM_BONUS', bonusTier: tier, amount: bonusPerTier, }); distributedBonus += bonusPerTier; } else { await createUnallocatedContribution(tx, { type: `BONUS_TIER_${tier}`, wouldBeAccount: directReferrer.accountSequence, amount: bonusPerTier, }); } } } } ``` ### 4.3 CDC 数据同步 ```typescript /** * Debezium CDC Consumer - 用户数据同步 */ @Consumer({ topic: 'dbserver1.rwa_identity.users' }) async handleUserCdc(message: DebeziumMessage): Promise { const { op, after, source } = message; const sequenceNum = source.sequence; // 幂等性检查 if (await isEventProcessed(`user-cdc-${sequenceNum}`)) { return; } switch (op) { case 'c': // CREATE case 'u': // UPDATE await syncedUserRepository.upsert({ accountSequence: after.account_sequence, originalUserId: after.id, phone: after.phone, status: after.status, sourceSequenceNum: sequenceNum, }); break; case 'd': // DELETE // 通常不处理删除,或标记为 inactive break; } await markEventProcessed(`user-cdc-${sequenceNum}`); } ``` --- ## 5. 服务间通信 ### 5.1 事件发布(Outbox Pattern) ```typescript // 发布算力计算完成事件 interface ContributionCalculatedEvent { eventId: string; eventType: 'ContributionCalculated'; accountSequence: string; totalContribution: string; effectiveContribution: string; calculatedAt: string; } // Mining Service 订阅此事件用于挖矿分配 ``` ### 5.2 订阅的 CDC Topics | Topic | 来源服务 | 数据内容 | |-------|---------|---------| | `dbserver1.rwa_identity.users` | identity-service | 用户基本信息 | | `dbserver1.rwa_planting.adoptions` | planting-service | 认种记录 | | `dbserver1.rwa_referral.referral_relations` | referral-service | 推荐关系 | --- ## 6. 关键注意事项 ### 6.1 数据一致性 - 所有算力变更必须在事务中完成 - 使用乐观锁 (version 字段) 处理并发更新 - 明细账与汇总账必须保持一致 ### 6.2 幂等性 - CDC 消息可能重复,使用 sequence_num 去重 - 认种分配使用 contribution_distributed 标记防止重复 ### 6.3 性能优化 - 团队统计表 (user_team_stats) 定期预计算 - 使用 Redis 缓存热点用户算力 - 批量处理历史数据计算 ### 6.4 跨服务关联 - **始终使用 account_sequence,不使用 userId** - account_sequence 是唯一的跨服务关联标识 --- ## 7. 开发检查清单 - [ ] 实现 CDC Consumer 同步用户/认种/推荐数据 - [ ] 实现算力计算核心逻辑 - [ ] 实现算力明细账记录 - [ ] 实现未分配算力归总部逻辑 - [ ] 实现算力过期处理定时任务 - [ ] 实现每日快照生成 - [ ] 实现查询 API - [ ] 编写单元测试 - [ ] 编写集成测试 - [ ] 配置 Debezium Connector --- ## 8. 启动命令 ```bash # 开发环境 npm run start:dev # 生成 Prisma Client npx prisma generate # 运行迁移 npx prisma migrate dev # 生产环境 npm run build && npm run start:prod ```