kaezhou/muyu-apiserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'feat: 多租户 CRM + PostgreSQL 图关系 + 租户隔离 + Casbin RBAC' (#1) from feat/multitenant-crm into main

Browse Source 
Reviewed-on: https://forgejo.cheverjohn.me/Business/muyuqingfeng-apiserver/pulls/1
This commit is contained in:
Chenwei Jiang 2026-03-30 21:01:30 +08:00
commit 9e80cdce0b
60 changed files with 2020 additions and 94 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -88,3 +88,5 @@ data/
.env
.env.*
*.pem
deploy/bin/
worker/graphsync/go.sum

8
deploy/Dockerfile.graphsync Normal file
View File

@ -0,0 +1,8 @@
FROM alpine:3.19
RUN apk add --no-cache ca-certificates tzdata
WORKDIR /app
COPY deploy/bin/graphsync .
CMD ["./graphsync"]

35
deploy/README-graph-stack.md Normal file
View File

@ -0,0 +1,35 @@
# Graph Stack Deploy Guide
This stack is the long-term CRM data architecture baseline:
- PostgreSQL as source of truth
- Kafka + Debezium for CDC
- Neo4j as graph analytics store
## Start
```bash
cd muyu-apiserver/deploy
docker compose -f docker-compose.graph.yaml up -d
```
## Endpoints
- PostgreSQL: `localhost:5432`
- Kafka broker: `localhost:9092`
- Debezium Connect: `http://localhost:8083`
- Neo4j Browser: `http://localhost:7474`
- Kafka UI: `http://localhost:8086`
## Stop
```bash
cd muyu-apiserver/deploy
docker compose -f docker-compose.graph.yaml down
```
## Notes
- This stack does not yet migrate existing MySQL business tables.
- Existing API services still run with current MySQL model code.
- Next phase is to add GraphSyncWorker and migrate APIs to PostgreSQL-backed CRM tables.

120
deploy/docker-compose.graph.yaml Normal file
View File

@ -0,0 +1,120 @@
services:
postgres:
image: postgres:16
container_name: muyu-postgres
restart: always
environment:
POSTGRES_USER: muyu
POSTGRES_PASSWORD: muyu2026
POSTGRES_DB: muyu_crm
TZ: Asia/Shanghai
ports:
- "5432:5432"
volumes:
- ./postgres/init.sql:/docker-entrypoint-initdb.d/init.sql
- ../data/postgres:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U muyu -d muyu_crm"]
interval: 10s
timeout: 5s
retries: 8
kafka:
image: bitnami/kafka:3.8
container_name: muyu-kafka
restart: always
environment:
- KAFKA_CFG_NODE_ID=1
- KAFKA_CFG_PROCESS_ROLES=broker,controller
- KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
- KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
- KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
- KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
- KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
- KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
- KAFKA_CFG_NUM_PARTITIONS=3
ports:
- "9092:9092"
volumes:
- ../data/kafka:/bitnami/kafka
healthcheck:
test: ["CMD-SHELL", "kafka-topics.sh --bootstrap-server localhost:9092 --list || exit 1"]
interval: 15s
timeout: 10s
retries: 10
debezium:
image: debezium/connect:2.7
container_name: muyu-debezium
restart: always
depends_on:
postgres:
condition: service_healthy
kafka:
condition: service_healthy
environment:
BOOTSTRAP_SERVERS: kafka:9092
GROUP_ID: 1
CONFIG_STORAGE_TOPIC: debezium_connect_configs
OFFSET_STORAGE_TOPIC: debezium_connect_offsets
STATUS_STORAGE_TOPIC: debezium_connect_statuses
CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
ports:
- "8083:8083"
neo4j:
image: neo4j:5.26
container_name: muyu-neo4j
restart: always
environment:
NEO4J_AUTH: neo4j/muyu2026
NEO4J_server_memory_heap_initial__size: 512m
NEO4J_server_memory_heap_max__size: 1024m
NEO4J_server_default__listen__address: 0.0.0.0
ports:
- "7474:7474"
- "7687:7687"
volumes:
- ../data/neo4j/data:/data
- ../data/neo4j/logs:/logs
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:7474 || exit 1"]
interval: 15s
timeout: 10s
retries: 10
graphsync:
build:
context: ..
dockerfile: deploy/Dockerfile.graphsync
container_name: muyu-graphsync
restart: always
depends_on:
kafka:
condition: service_healthy
neo4j:
condition: service_healthy
environment:
KAFKA_BROKERS: kafka:9092
KAFKA_TOPIC: muyu_crm.public.crm_tenant_relationship
KAFKA_GROUP: graphsync
NEO4J_URI: bolt://neo4j:7687
NEO4J_USER: neo4j
NEO4J_PASS: muyu2026
TZ: Asia/Shanghai
kafka-ui:
image: provectuslabs/kafka-ui:latest
container_name: muyu-kafka-ui
restart: always
depends_on:
kafka:
condition: service_healthy
environment:
KAFKA_CLUSTERS_0_NAME: local
KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
ports:
- "8086:8080"

21
deploy/docker-compose.yaml
View File

@ -33,6 +33,26 @@ services:
timeout: 5s
retries: 5
postgres:
image: postgres:16
container_name: muyu-postgres
restart: always
environment:
POSTGRES_USER: muyu
POSTGRES_PASSWORD: muyu2026
POSTGRES_DB: muyu_crm
TZ: Asia/Shanghai
ports:
- "5432:5432"
volumes:
- ./postgres/init-crm.sql:/docker-entrypoint-initdb.d/init.sql
- ../data/postgres:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U muyu -d muyu_crm"]
interval: 10s
timeout: 5s
retries: 8
etcd:
image: quay.io/coreos/etcd:v3.5.17
container_name: muyu-etcd
@ -98,6 +118,7 @@ services:
depends_on:
- system-rpc
- inventory-rpc
- postgres
environment:
TZ: Asia/Shanghai

4
deploy/etc/gateway.yaml
View File

@ -5,6 +5,10 @@ Port: 8888
Auth:
AccessSecret: muyu-wms-jwt-secret-key-2026
AccessExpire: 7200
DefaultTenantId: t_default_001
DataSource: root:muyu2026@tcp(mysql:3306)/muyu_wms?charset=utf8mb4&parseTime=true&loc=Asia%2FShanghai
PostgresDataSource: "postgres://muyu:muyu2026@postgres:5432/muyu_crm?sslmode=disable"
SystemRpc:
Etcd:

101
deploy/mysql/init.sql
View File

@ -89,6 +89,8 @@ CREATE TABLE `sys_operation_log` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`log_id` VARCHAR(32) NOT NULL,
`user_id` VARCHAR(32) NOT NULL DEFAULT '',
`tenant_id` VARCHAR(32) NOT NULL DEFAULT '',
`target_tenant_id` VARCHAR(32) NOT NULL DEFAULT '',
`username` VARCHAR(50) NOT NULL DEFAULT '',
`module` VARCHAR(50) NOT NULL DEFAULT '',
`operation` VARCHAR(20) NOT NULL DEFAULT '',
@ -104,6 +106,7 @@ CREATE TABLE `sys_operation_log` (
PRIMARY KEY (`id`),
UNIQUE KEY `uk_log_id` (`log_id`),
KEY `idx_user_id` (`user_id`),
KEY `idx_tenant_id` (`tenant_id`),
KEY `idx_module` (`module`),
KEY `idx_created_at` (`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
@ -124,6 +127,85 @@ CREATE TABLE `sys_config` (
KEY `idx_config_group` (`config_group`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- ==================== CRM Multi-tenant and Graph Tables ====================
CREATE TABLE `crm_tenant` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`tenant_id` VARCHAR(32) NOT NULL,
`tenant_name` VARCHAR(100) NOT NULL,
`tenant_code` VARCHAR(64) NOT NULL DEFAULT '',
`status` TINYINT NOT NULL DEFAULT 1 COMMENT '1:enabled 0:disabled',
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tenant_id` (`tenant_id`),
UNIQUE KEY `uk_tenant_code` (`tenant_code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE `crm_tenant_user` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`tenant_user_id` VARCHAR(32) NOT NULL,
`tenant_id` VARCHAR(32) NOT NULL,
`user_id` VARCHAR(32) NOT NULL,
`is_primary` TINYINT NOT NULL DEFAULT 1 COMMENT '1:primary 0:secondary',
`status` TINYINT NOT NULL DEFAULT 1 COMMENT '1:enabled 0:disabled',
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tenant_user_id` (`tenant_user_id`),
UNIQUE KEY `uk_tenant_user_pair` (`tenant_id`, `user_id`),
KEY `idx_user_id` (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE `crm_tenant_relationship` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`relationship_id` VARCHAR(40) NOT NULL,
`owner_tenant_id` VARCHAR(32) NOT NULL,
`from_tenant_id` VARCHAR(32) NOT NULL,
`to_tenant_id` VARCHAR(32) NOT NULL,
`relation_type` VARCHAR(32) NOT NULL,
`status` TINYINT NOT NULL DEFAULT 1 COMMENT '1:active 0:inactive',
`valid_from` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`valid_to` DATETIME DEFAULT NULL,
`source` VARCHAR(32) NOT NULL DEFAULT 'manual',
`created_by` VARCHAR(32) NOT NULL DEFAULT '',
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_relationship_id` (`relationship_id`),
UNIQUE KEY `uk_relationship_unique` (`owner_tenant_id`, `from_tenant_id`, `to_tenant_id`, `relation_type`, `valid_from`),
KEY `idx_owner_from_status` (`owner_tenant_id`, `from_tenant_id`, `status`),
KEY `idx_owner_to_status` (`owner_tenant_id`, `to_tenant_id`, `status`),
CONSTRAINT `chk_no_self_loop` CHECK (`from_tenant_id` <> `to_tenant_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE `crm_tenant_relationship_history` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`history_id` VARCHAR(40) NOT NULL,
`relationship_id` VARCHAR(40) NOT NULL,
`change_type` VARCHAR(32) NOT NULL,
`before_json` JSON NULL,
`after_json` JSON NULL,
`changed_by` VARCHAR(32) NOT NULL DEFAULT '',
`changed_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_history_id` (`history_id`),
KEY `idx_relationship_id` (`relationship_id`),
KEY `idx_changed_at` (`changed_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE `crm_data_migration_log` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`migration_id` VARCHAR(40) NOT NULL,
`migration_name` VARCHAR(100) NOT NULL,
`status` TINYINT NOT NULL DEFAULT 0 COMMENT '0:running 1:success 2:failed',
`detail` TEXT,
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_migration_id` (`migration_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
-- ==================== Inventory Management Tables ====================
CREATE TABLE `inv_product` (
@ -146,7 +228,7 @@ CREATE TABLE `inv_product` (
`deleted_at` DATETIME DEFAULT NULL,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_product_id` (`product_id`),
UNIQUE KEY `uk_product_name` (`product_name`),
UNIQUE KEY `uk_product_name_spec_color` (`product_name`, `spec`, `color`),
KEY `idx_color` (`color`),
KEY `idx_location` (`location`),
KEY `idx_status` (`status`),
@ -255,6 +337,13 @@ CREATE TABLE `casbin_rule` (
INSERT INTO `sys_user` (`user_id`, `username`, `password`, `salt`, `real_name`, `status`)
VALUES ('u_admin_001', 'admin', '$2a$10$S5XPWRXUnexAt9KJw2CefeHMa6aude7j5dm5qgWM1YMQAAwRgsGXa', 'bcrypt', 'System Admin', 1);
-- Default tenant and mapping
INSERT INTO `crm_tenant` (`tenant_id`, `tenant_name`, `tenant_code`, `status`)
VALUES ('t_default_001', '默认租户', 'DEFAULT', 1);
INSERT INTO `crm_tenant_user` (`tenant_user_id`, `tenant_id`, `user_id`, `is_primary`, `status`)
VALUES ('tu_001', 't_default_001', 'u_admin_001', 1, 1);
-- Default roles
INSERT INTO `sys_role` (`role_id`, `role_name`, `role_key`, `role_desc`, `sort_order`, `status`) VALUES
('r_001', '系统管理员', 'admin', 'Full system access', 1, 1),
@ -284,6 +373,11 @@ INSERT INTO `sys_menu` (`menu_id`, `parent_id`, `menu_name`, `menu_type`, `path`
('m_205', 'm_200', '库存调整', 2, '/inventory/adjusts', './Inventory/Adjusts', 'inventory:adjust:list', 'SwapOutlined', 5, 1, 1),
('m_206', 'm_200', 'Excel导入', 2, '/inventory/import', './Inventory/Import', 'inventory:import:exec', 'ImportOutlined', 6, 1, 1);
-- CRM menus
INSERT INTO `sys_menu` (`menu_id`, `parent_id`, `menu_name`, `menu_type`, `path`, `component`, `permission`, `icon`, `sort_order`, `visible`, `status`) VALUES
('m_300', '', '客户关系', 1, '/crm', '', '', 'NodeIndexOutlined', 300, 1, 1),
('m_301', 'm_300', '上下游关系', 2, '/crm/relations', './CRM/Relations', 'crm:relation:list', 'ApartmentOutlined', 1, 1, 1);
-- Assign all menus to admin role
INSERT INTO `sys_role_menu` (`role_id`, `menu_id`)
SELECT 'r_001', `menu_id` FROM `sys_menu`;
@ -294,11 +388,16 @@ INSERT INTO `sys_role_menu` (`role_id`, `menu_id`) VALUES
('r_002', 'm_200'), ('r_002', 'm_201'), ('r_002', 'm_202'),
('r_002', 'm_203'), ('r_002', 'm_204'), ('r_002', 'm_205'), ('r_002', 'm_206');
-- Assign CRM menus to admin role
INSERT INTO `sys_role_menu` (`role_id`, `menu_id`) VALUES
('r_001', 'm_300'), ('r_001', 'm_301');
-- Default Casbin policies
INSERT INTO `casbin_rule` (`ptype`, `v0`, `v1`, `v2`) VALUES
('p', 'admin', '/api/v1/*', '*'),
('p', 'warehouse', '/api/v1/auth/*', '*'),
('p', 'warehouse', '/api/v1/inventory/*', '*'),
('p', 'admin', '/api/v1/crm/*', '*'),
('p', 'user', '/api/v1/auth/*', '*'),
('p', 'user', '/api/v1/inventory/products', 'GET'),
('p', 'user', '/api/v1/inventory/stocks', 'GET'),

93
deploy/mysql/migrations/20260329_add_tenant_id_to_legacy.sql Normal file
View File

@ -0,0 +1,93 @@
-- Migration: Add tenant_id to all legacy tables
-- Safe to re-run: uses column existence checks via PREPARE/EXECUTE
USE `muyu_wms`;
-- Helper: add tenant_id column if not exists
-- Repeated for each table because MySQL 8 lacks ADD COLUMN IF NOT EXISTS.
-- sys_user
SET @t = 'sys_user';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- sys_role
SET @t = 'sys_role';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- sys_user_role
SET @t = 'sys_user_role';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- sys_menu
SET @t = 'sys_menu';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- sys_role_menu
SET @t = 'sys_role_menu';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- sys_config
SET @t = 'sys_config';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- inv_product
SET @t = 'inv_product';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- inv_stock_check
SET @t = 'inv_stock_check';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- inv_stock_check_detail
SET @t = 'inv_stock_check_detail';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- inv_stock_adjust
SET @t = 'inv_stock_adjust';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- inv_stock_adjust_detail
SET @t = 'inv_stock_adjust_detail';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- inv_stock_import_log
SET @t = 'inv_stock_import_log';
SET @c = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME=@t AND COLUMN_NAME='tenant_id');
SET @s = IF(@c=0, CONCAT('ALTER TABLE `',@t,'` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'t_default_001\' AFTER `id`'), 'SELECT 1');
PREPARE stmt FROM @s; EXECUTE stmt; DEALLOCATE PREPARE stmt;
-- Add indexes for tenant_id on all tables
CREATE INDEX IF NOT EXISTS idx_sys_user_tenant ON sys_user(tenant_id);
CREATE INDEX IF NOT EXISTS idx_sys_role_tenant ON sys_role(tenant_id);
CREATE INDEX IF NOT EXISTS idx_sys_user_role_tenant ON sys_user_role(tenant_id);
CREATE INDEX IF NOT EXISTS idx_sys_menu_tenant ON sys_menu(tenant_id);
CREATE INDEX IF NOT EXISTS idx_sys_role_menu_tenant ON sys_role_menu(tenant_id);
CREATE INDEX IF NOT EXISTS idx_sys_config_tenant ON sys_config(tenant_id);
CREATE INDEX IF NOT EXISTS idx_inv_product_tenant ON inv_product(tenant_id);
CREATE INDEX IF NOT EXISTS idx_inv_stock_check_tenant ON inv_stock_check(tenant_id);
CREATE INDEX IF NOT EXISTS idx_inv_stock_check_detail_tenant ON inv_stock_check_detail(tenant_id);
CREATE INDEX IF NOT EXISTS idx_inv_stock_adjust_tenant ON inv_stock_adjust(tenant_id);
CREATE INDEX IF NOT EXISTS idx_inv_stock_adjust_detail_tenant ON inv_stock_adjust_detail(tenant_id);
CREATE INDEX IF NOT EXISTS idx_inv_stock_import_log_tenant ON inv_stock_import_log(tenant_id);

87
deploy/mysql/migrations/20260329_multitenant_crm.sql Normal file
View File

@ -0,0 +1,87 @@
-- Incremental migration: multi-tenant CRM baseline
-- Apply to existing muyu_wms schema.
USE `muyu_wms`;
-- MySQL 8 does not support ADD COLUMN IF NOT EXISTS; use separate statements.
SET @col_exists = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME='sys_operation_log' AND COLUMN_NAME='tenant_id');
SET @sql = IF(@col_exists = 0, 'ALTER TABLE `sys_operation_log` ADD COLUMN `tenant_id` VARCHAR(32) NOT NULL DEFAULT \'\' AFTER `user_id`', 'SELECT 1');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
SET @col_exists = (SELECT COUNT(*) FROM information_schema.COLUMNS WHERE TABLE_SCHEMA='muyu_wms' AND TABLE_NAME='sys_operation_log' AND COLUMN_NAME='target_tenant_id');
SET @sql = IF(@col_exists = 0, 'ALTER TABLE `sys_operation_log` ADD COLUMN `target_tenant_id` VARCHAR(32) NOT NULL DEFAULT \'\' AFTER `tenant_id`', 'SELECT 1');
PREPARE stmt FROM @sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;
CREATE TABLE IF NOT EXISTS `crm_tenant` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`tenant_id` VARCHAR(32) NOT NULL,
`tenant_name` VARCHAR(100) NOT NULL,
`tenant_code` VARCHAR(64) NOT NULL DEFAULT '',
`status` TINYINT NOT NULL DEFAULT 1 COMMENT '1:enabled 0:disabled',
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tenant_id` (`tenant_id`),
UNIQUE KEY `uk_tenant_code` (`tenant_code`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE IF NOT EXISTS `crm_tenant_user` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`tenant_user_id` VARCHAR(32) NOT NULL,
`tenant_id` VARCHAR(32) NOT NULL,
`user_id` VARCHAR(32) NOT NULL,
`is_primary` TINYINT NOT NULL DEFAULT 1 COMMENT '1:primary 0:secondary',
`status` TINYINT NOT NULL DEFAULT 1 COMMENT '1:enabled 0:disabled',
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_tenant_user_id` (`tenant_user_id`),
UNIQUE KEY `uk_tenant_user_pair` (`tenant_id`, `user_id`),
KEY `idx_user_id` (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE IF NOT EXISTS `crm_tenant_relationship` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`relationship_id` VARCHAR(40) NOT NULL,
`owner_tenant_id` VARCHAR(32) NOT NULL,
`from_tenant_id` VARCHAR(32) NOT NULL,
`to_tenant_id` VARCHAR(32) NOT NULL,
`relation_type` VARCHAR(32) NOT NULL,
`status` TINYINT NOT NULL DEFAULT 1 COMMENT '1:active 0:inactive',
`valid_from` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`valid_to` DATETIME DEFAULT NULL,
`source` VARCHAR(32) NOT NULL DEFAULT 'manual',
`created_by` VARCHAR(32) NOT NULL DEFAULT '',
`created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
`updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_relationship_id` (`relationship_id`),
UNIQUE KEY `uk_relationship_unique` (`owner_tenant_id`, `from_tenant_id`, `to_tenant_id`, `relation_type`, `valid_from`),
KEY `idx_owner_from_status` (`owner_tenant_id`, `from_tenant_id`, `status`),
KEY `idx_owner_to_status` (`owner_tenant_id`, `to_tenant_id`, `status`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
CREATE TABLE IF NOT EXISTS `crm_tenant_relationship_history` (
`id` BIGINT NOT NULL AUTO_INCREMENT,
`history_id` VARCHAR(40) NOT NULL,
`relationship_id` VARCHAR(40) NOT NULL,
`change_type` VARCHAR(32) NOT NULL,
`before_json` JSON NULL,
`after_json` JSON NULL,
`changed_by` VARCHAR(32) NOT NULL DEFAULT '',
`changed_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_history_id` (`history_id`),
KEY `idx_relationship_id` (`relationship_id`),
KEY `idx_changed_at` (`changed_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
INSERT INTO `crm_tenant` (`tenant_id`, `tenant_name`, `tenant_code`, `status`)
VALUES ('t_default_001', '默认租户', 'DEFAULT', 1)
ON DUPLICATE KEY UPDATE tenant_name = VALUES(tenant_name), status = VALUES(status);
INSERT INTO `crm_tenant_user` (`tenant_user_id`, `tenant_id`, `user_id`, `is_primary`, `status`)
SELECT 'tu_admin_001', 't_default_001', 'u_admin_001', 1, 1
WHERE NOT EXISTS (
SELECT 1 FROM `crm_tenant_user` WHERE `tenant_id` = 't_default_001' AND `user_id` = 'u_admin_001'
);

5
deploy/nginx/nginx.conf
View File

@ -1,6 +1,7 @@
server {
listen 80;
server_name localhost;
resolver 127.0.0.11 valid=30s ipv6=off;
root /usr/share/nginx/html;
index index.html;
@ -12,7 +13,9 @@ server {
# API proxy
location /api/ {
proxy_pass http://gateway:8888;
# Use Docker DNS dynamic resolution to avoid stale container IP after recreate.
set $gateway_upstream http://gateway:8888;
proxy_pass $gateway_upstream;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

70
deploy/postgres/init-crm.sql Normal file
View File

@ -0,0 +1,70 @@
-- CRM tables in PostgreSQL (mirrors MySQL crm_* schema for CrmRepo compatibility)
CREATE TABLE IF NOT EXISTS crm_tenant (
id BIGSERIAL PRIMARY KEY,
tenant_id VARCHAR(32) NOT NULL UNIQUE,
tenant_name VARCHAR(100) NOT NULL,
tenant_code VARCHAR(64) NOT NULL DEFAULT '' UNIQUE,
status SMALLINT NOT NULL DEFAULT 1,
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS crm_tenant_user (
id BIGSERIAL PRIMARY KEY,
tenant_user_id VARCHAR(32) NOT NULL UNIQUE,
tenant_id VARCHAR(32) NOT NULL,
user_id VARCHAR(32) NOT NULL,
is_primary SMALLINT NOT NULL DEFAULT 1,
status SMALLINT NOT NULL DEFAULT 1,
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
UNIQUE (tenant_id, user_id)
);
CREATE INDEX IF NOT EXISTS idx_crm_tenant_user_uid ON crm_tenant_user(user_id);
CREATE TABLE IF NOT EXISTS crm_tenant_relationship (
id BIGSERIAL PRIMARY KEY,
relationship_id VARCHAR(40) NOT NULL UNIQUE,
owner_tenant_id VARCHAR(32) NOT NULL,
from_tenant_id VARCHAR(32) NOT NULL,
to_tenant_id VARCHAR(32) NOT NULL,
relation_type VARCHAR(32) NOT NULL,
status SMALLINT NOT NULL DEFAULT 1,
valid_from TIMESTAMP NOT NULL DEFAULT NOW(),
valid_to TIMESTAMP NULL,
source VARCHAR(32) NOT NULL DEFAULT 'manual',
created_by VARCHAR(32) NOT NULL DEFAULT '',
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
updated_at TIMESTAMP NOT NULL DEFAULT NOW(),
CHECK (from_tenant_id <> to_tenant_id)
);
CREATE UNIQUE INDEX IF NOT EXISTS uk_crm_rel_unique
ON crm_tenant_relationship(owner_tenant_id, from_tenant_id, to_tenant_id, relation_type, valid_from);
CREATE INDEX IF NOT EXISTS idx_crm_rel_from ON crm_tenant_relationship(owner_tenant_id, from_tenant_id, status);
CREATE INDEX IF NOT EXISTS idx_crm_rel_to ON crm_tenant_relationship(owner_tenant_id, to_tenant_id, status);
CREATE TABLE IF NOT EXISTS crm_tenant_relationship_history (
id BIGSERIAL PRIMARY KEY,
history_id VARCHAR(40) NOT NULL UNIQUE,
relationship_id VARCHAR(40) NOT NULL,
change_type VARCHAR(32) NOT NULL,
before_json JSONB NULL,
after_json JSONB NULL,
changed_by VARCHAR(32) NOT NULL DEFAULT '',
changed_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_crm_rel_hist ON crm_tenant_relationship_history(relationship_id, changed_at DESC);
-- Seed
INSERT INTO crm_tenant (tenant_id, tenant_name, tenant_code, status)
VALUES ('t_default_001', '默认租户', 'DEFAULT', 1)
ON CONFLICT (tenant_id) DO NOTHING;
INSERT INTO crm_tenant_user (tenant_user_id, tenant_id, user_id, is_primary, status)
VALUES ('tu_admin_001', 't_default_001', 'u_admin_001', 1, 1)
ON CONFLICT (tenant_id, user_id) DO NOTHING;

105
deploy/postgres/init.sql Normal file
View File

@ -0,0 +1,105 @@
-- Muyu CRM Graph Schema (PostgreSQL)
-- Long-term architecture baseline:
-- PostgreSQL (source of truth) -> CDC/Kafka -> Neo4j (graph analytics)
CREATE SCHEMA IF NOT EXISTS crm;
SET search_path TO crm, public;
-- ==================== Tenant and Identity ====================
CREATE TABLE IF NOT EXISTS tenant (
id BIGSERIAL PRIMARY KEY,
tenant_code VARCHAR(64) NOT NULL UNIQUE,
tenant_name VARCHAR(128) NOT NULL,
status SMALLINT NOT NULL DEFAULT 1,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS tenant_user (
id BIGSERIAL PRIMARY KEY,
tenant_id BIGINT NOT NULL REFERENCES tenant(id),
user_id VARCHAR(64) NOT NULL,
role_key VARCHAR(64) NOT NULL DEFAULT 'user',
is_primary BOOLEAN NOT NULL DEFAULT TRUE,
status SMALLINT NOT NULL DEFAULT 1,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (tenant_id, user_id)
);
CREATE INDEX IF NOT EXISTS idx_tenant_user_user_id ON tenant_user(user_id);
-- ==================== Graph Nodes and Edges ====================
CREATE TABLE IF NOT EXISTS relationship_type (
id BIGSERIAL PRIMARY KEY,
type_code VARCHAR(32) NOT NULL UNIQUE,
type_name VARCHAR(64) NOT NULL,
description TEXT NOT NULL DEFAULT '',
status SMALLINT NOT NULL DEFAULT 1,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TABLE IF NOT EXISTS tenant_relationship (
id BIGSERIAL PRIMARY KEY,
owner_tenant_id BIGINT NOT NULL REFERENCES tenant(id),
from_tenant_id BIGINT NOT NULL REFERENCES tenant(id),
to_tenant_id BIGINT NOT NULL REFERENCES tenant(id),
relationship_type_id BIGINT NOT NULL REFERENCES relationship_type(id),
status SMALLINT NOT NULL DEFAULT 1,
valid_from TIMESTAMPTZ NOT NULL DEFAULT NOW(),
valid_to TIMESTAMPTZ NULL,
source VARCHAR(32) NOT NULL DEFAULT 'manual',
created_by VARCHAR(64) NOT NULL DEFAULT 'system',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CHECK (from_tenant_id <> to_tenant_id),
CHECK (valid_to IS NULL OR valid_to > valid_from)
);
CREATE UNIQUE INDEX IF NOT EXISTS uk_tenant_relationship_version
ON tenant_relationship(owner_tenant_id, from_tenant_id, to_tenant_id, relationship_type_id, valid_from);
CREATE INDEX IF NOT EXISTS idx_tenant_relationship_from
ON tenant_relationship(owner_tenant_id, from_tenant_id, status);
CREATE INDEX IF NOT EXISTS idx_tenant_relationship_to
ON tenant_relationship(owner_tenant_id, to_tenant_id, status);
CREATE TABLE IF NOT EXISTS tenant_relationship_history (
id BIGSERIAL PRIMARY KEY,
relationship_id BIGINT NOT NULL REFERENCES tenant_relationship(id),
change_type VARCHAR(32) NOT NULL,
before_json JSONB,
after_json JSONB,
changed_by VARCHAR(64) NOT NULL DEFAULT 'system',
changed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_tenant_relationship_history_rel
ON tenant_relationship_history(relationship_id, changed_at DESC);
-- ==================== Seed Data ====================
INSERT INTO relationship_type(type_code, type_name, description)
VALUES
('SUPPLY', '供货关系', 'from_tenant 作为 to_tenant 的上游'),
('PURCHASE', '采购关系', 'from_tenant 向 to_tenant 发起采购'),
('OEM', '代工关系', '委托代工上下游关系')
ON CONFLICT (type_code) DO NOTHING;
-- ==================== One-hop Query Examples ====================
-- Upstream (who supplies current tenant):
-- SELECT r.from_tenant_id
-- FROM crm.tenant_relationship r
-- WHERE r.owner_tenant_id = $1
-- AND r.to_tenant_id = $2
-- AND r.status = 1;
--
-- Downstream (who current tenant supplies):
-- SELECT r.to_tenant_id
-- FROM crm.tenant_relationship r
-- WHERE r.owner_tenant_id = $1
-- AND r.from_tenant_id = $2
-- AND r.status = 1;

7
gateway/etc/gateway-api.yaml
View File

@ -1,3 +1,10 @@
Name: gateway-api
Host: 0.0.0.0
Port: 8888
Auth:
AccessSecret: muyu-wms-jwt-secret-key-2026
AccessExpire: 7200
DefaultTenantId: t_default_001
DataSource: root:muyu2026@tcp(127.0.0.1:3306)/muyu_wms?charset=utf8mb4&parseTime=true&loc=Asia%2FShanghai

4
gateway/etc/gateway.yaml
View File

@ -5,6 +5,10 @@ Port: 8888
Auth:
AccessSecret: muyu-wms-jwt-secret-key-2026
AccessExpire: 7200
DefaultTenantId: t_default_001
DataSource: root:muyu2026@tcp(127.0.0.1:3306)/muyu_wms?charset=utf8mb4&parseTime=true&loc=Asia%2FShanghai
PostgresDataSource: "postgres://muyu:muyu2026@127.0.0.1:5432/muyu_crm?sslmode=disable"
SystemRpc:
Etcd:

76
gateway/gateway.api
View File

@ -22,6 +22,7 @@ type (
LoginReq {
Username string `json:"username"`
Password string `json:"password"`
TenantId string `json:"tenantId,optional"`
}
LoginResp {
Token string `json:"token"`
@ -45,6 +46,7 @@ type (
Status int64 `json:"status"`
RoleId string `json:"roleId"`
RoleName string `json:"roleName"`
TenantId string `json:"tenantId"`
LastLoginTime string `json:"lastLoginTime"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
@ -407,6 +409,56 @@ type (
}
)
// ==================== CRM Relation Types ====================
type (
CrmRelationInfo {
RelationshipId string `json:"relationshipId"`
FromTenantId string `json:"fromTenantId"`
FromTenantName string `json:"fromTenantName"`
ToTenantId string `json:"toTenantId"`
ToTenantName string `json:"toTenantName"`
RelationType string `json:"relationType"`
Status int64 `json:"status"`
ValidFrom string `json:"validFrom"`
ValidTo string `json:"validTo,optional"`
}
CrmListRelationsReq {
Depth int64 `form:"depth,default=1"`
}
CrmListRelationsResp {
Total int64 `json:"total"`
List []CrmRelationInfo `json:"list"`
}
CrmCreateRelationReq {
FromTenantId string `json:"fromTenantId"`
ToTenantId string `json:"toTenantId"`
RelationType string `json:"relationType"`
ValidFrom string `json:"validFrom,optional"`
ValidTo string `json:"validTo,optional"`
}
CrmUpdateRelationReq {
Status int64 `json:"status"`
ValidFrom string `json:"validFrom,optional"`
ValidTo string `json:"validTo,optional"`
}
CrmRelationHistoryReq {
Page int64 `form:"page,default=1"`
PageSize int64 `form:"pageSize,default=20"`
}
CrmRelationHistoryItem {
RelationshipId string `json:"relationshipId"`
ChangeType string `json:"changeType"`
BeforeJson string `json:"beforeJson"`
AfterJson string `json:"afterJson"`
ChangedBy string `json:"changedBy"`
ChangedAt string `json:"changedAt"`
}
CrmRelationHistoryResp {
Total int64 `json:"total"`
List []CrmRelationHistoryItem `json:"list"`
}
)
// ==================== Services ====================
// Auth - No JWT required
@server (
@ -418,6 +470,30 @@ service gateway-api {
post /login (LoginReq) returns (LoginResp)
}
// CRM - Graph & Relations
@server (
prefix: /api/v1/crm
group: crm/relation
jwt: Auth
middleware: AuthorityMiddleware
)
service gateway-api {
@handler ListUpstreamHandler
get /graph/upstream (CrmListRelationsReq) returns (CrmListRelationsResp)
@handler ListDownstreamHandler
get /graph/downstream (CrmListRelationsReq) returns (CrmListRelationsResp)
@handler CreateRelationHandler
post /relationships (CrmCreateRelationReq) returns (IdResp)
@handler UpdateRelationHandler
put /relationships/:id (CrmUpdateRelationReq) returns (IdResp)
@handler ListRelationHistoryHandler
get /relationships/history (CrmRelationHistoryReq) returns (CrmRelationHistoryResp)
}
// Auth - JWT required
@server (
prefix: /api/v1/auth

9
gateway/gateway.go
View File

@ -36,6 +36,15 @@ func main() {
server := rest.MustNewServer(c.RestConf, rest.WithCors())
defer server.Stop()
server.Use(func(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
if token := r.Header.Get("X-Token"); token != "" {
r.Header.Set("Authorization", "Bearer "+token)
}
next(w, r)
}
})
httpx.SetErrorHandlerCtx(func(ctx context.Context, err error) (int, interface{}) {
s, ok := status.FromError(err)
if ok {

3
gateway/internal/config/config.go
View File

@ -13,7 +13,10 @@ type Config struct {
Auth struct {
AccessSecret string
AccessExpire int64
DefaultTenantId string
}
DataSource string
PostgresDataSource string
SystemRpc zrpc.RpcClientConf
InventoryRpc zrpc.RpcClientConf
}

30
gateway/internal/handler/crm/relation/createRelationHandler.go Normal file
View File

@ -0,0 +1,30 @@
package relation
import (
"net/http"
"muyu-apiserver/gateway/internal/logic/crm/relation"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"github.com/zeromicro/go-zero/rest/httpx"
)
func CreateRelationHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req types.CrmCreateRelationReq
if err := httpx.Parse(r, &req); err != nil {
httpx.ErrorCtx(r.Context(), w, err)
return
}
l := relation.NewCreateRelationLogic(r.Context(), svcCtx)
resp, err := l.CreateRelation(&req)
if err != nil {
httpx.ErrorCtx(r.Context(), w, err)
} else {
httpx.OkJsonCtx(r.Context(), w, resp)
}
}
}

33
gateway/internal/handler/crm/relation/getFullGraphHandler.go Normal file
View File

@ -0,0 +1,33 @@
package relation
import (
"encoding/json"
"net/http"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/pkg/ctxdata"
"github.com/zeromicro/go-zero/rest/httpx"
)
func GetFullGraphHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
roleKey := ctxdata.GetRoleKey(r.Context())
if roleKey != "admin" {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusForbidden)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 1006,
"msg": "access denied: admin only",
})
return
}
graph, err := svcCtx.CrmRepo.GetFullGraph(r.Context())
if err != nil {
httpx.ErrorCtx(r.Context(), w, err)
return
}
httpx.OkJsonCtx(r.Context(), w, graph)
}
}

30
gateway/internal/handler/crm/relation/listDownstreamHandler.go Normal file
View File

@ -0,0 +1,30 @@
package relation
import (
"net/http"
"muyu-apiserver/gateway/internal/logic/crm/relation"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"github.com/zeromicro/go-zero/rest/httpx"
)
func ListDownstreamHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req types.CrmListRelationsReq
if err := httpx.Parse(r, &req); err != nil {
httpx.ErrorCtx(r.Context(), w, err)
return
}
l := relation.NewListDownstreamLogic(r.Context(), svcCtx)
resp, err := l.ListDownstream(&req)
if err != nil {
httpx.ErrorCtx(r.Context(), w, err)
} else {
httpx.OkJsonCtx(r.Context(), w, resp)
}
}
}

30
gateway/internal/handler/crm/relation/listRelationHistoryHandler.go Normal file
View File

@ -0,0 +1,30 @@
package relation
import (
"net/http"
"muyu-apiserver/gateway/internal/logic/crm/relation"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"github.com/zeromicro/go-zero/rest/httpx"
)
func ListRelationHistoryHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req types.CrmRelationHistoryReq
if err := httpx.Parse(r, &req); err != nil {
httpx.ErrorCtx(r.Context(), w, err)
return
}
l := relation.NewListRelationHistoryLogic(r.Context(), svcCtx)
resp, err := l.ListRelationHistory(&req)
if err != nil {
httpx.ErrorCtx(r.Context(), w, err)
} else {
httpx.OkJsonCtx(r.Context(), w, resp)
}
}
}

30
gateway/internal/handler/crm/relation/listUpstreamHandler.go Normal file
View File

@ -0,0 +1,30 @@
package relation
import (
"net/http"
"muyu-apiserver/gateway/internal/logic/crm/relation"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"github.com/zeromicro/go-zero/rest/httpx"
)
func ListUpstreamHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req types.CrmListRelationsReq
if err := httpx.Parse(r, &req); err != nil {
httpx.ErrorCtx(r.Context(), w, err)
return
}
l := relation.NewListUpstreamLogic(r.Context(), svcCtx)
resp, err := l.ListUpstream(&req)
if err != nil {
httpx.ErrorCtx(r.Context(), w, err)
} else {
httpx.OkJsonCtx(r.Context(), w, resp)
}
}
}

33
gateway/internal/handler/crm/relation/updateRelationHandler.go Normal file
View File

@ -0,0 +1,33 @@
package relation
import (
"context"
"net/http"
"muyu-apiserver/gateway/internal/logic/crm/relation"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"github.com/zeromicro/go-zero/rest/httpx"
"github.com/zeromicro/go-zero/rest/pathvar"
)
func UpdateRelationHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
var req types.CrmUpdateRelationReq
if err := httpx.Parse(r, &req); err != nil {
httpx.ErrorCtx(r.Context(), w, err)
return
}
ctx := context.WithValue(r.Context(), "pathId", pathvar.Vars(r)["id"])
l := relation.NewUpdateRelationLogic(ctx, svcCtx)
resp, err := l.UpdateRelation(&req)
if err != nil {
httpx.ErrorCtx(r.Context(), w, err)
} else {
httpx.OkJsonCtx(r.Context(), w, resp)
}
}
}

41
gateway/internal/handler/routes.go
View File

@ -7,6 +7,7 @@ import (
"net/http"
auth "muyu-apiserver/gateway/internal/handler/auth"
crmrelation "muyu-apiserver/gateway/internal/handler/crm/relation"
inventoryadjust "muyu-apiserver/gateway/internal/handler/inventory/adjust"
inventorycheck "muyu-apiserver/gateway/internal/handler/inventory/check"
inventoryproduct "muyu-apiserver/gateway/internal/handler/inventory/product"
@ -55,6 +56,46 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
rest.WithPrefix("/api/v1/auth"),
)
server.AddRoutes(
rest.WithMiddlewares(
[]rest.Middleware{serverCtx.AuthorityMiddleware},
[]rest.Route{
{
Method: http.MethodGet,
Path: "/graph/full",
Handler: crmrelation.GetFullGraphHandler(serverCtx),
},
{
Method: http.MethodGet,
Path: "/graph/upstream",
Handler: crmrelation.ListUpstreamHandler(serverCtx),
},
{
Method: http.MethodGet,
Path: "/graph/downstream",
Handler: crmrelation.ListDownstreamHandler(serverCtx),
},
{
Method: http.MethodPost,
Path: "/relationships",
Handler: crmrelation.CreateRelationHandler(serverCtx),
},
{
Method: http.MethodPut,
Path: "/relationships/:id",
Handler: crmrelation.UpdateRelationHandler(serverCtx),
},
{
Method: http.MethodGet,
Path: "/relationships/history",
Handler: crmrelation.ListRelationHistoryHandler(serverCtx),
},
}...,
),
rest.WithJwt(serverCtx.Config.Auth.AccessSecret),
rest.WithPrefix("/api/v1/crm"),
)
server.AddRoutes(
rest.WithMiddlewares(
[]rest.Middleware{serverCtx.AuthorityMiddleware},

2
gateway/internal/logic/auth/getUserInfoLogic.go
View File

@ -32,6 +32,7 @@ type UserInfoWithMenus struct {
func (l *GetUserInfoLogic) GetUserInfo() (*UserInfoWithMenus, error) {
userId := ctxdata.GetUserId(l.ctx)
tenantId := ctxdata.GetTenantId(l.ctx)
rpcResp, err := l.svcCtx.SystemRpc.GetUser(l.ctx, &pb.GetUserReq{
UserId: userId,
@ -71,6 +72,7 @@ func (l *GetUserInfoLogic) GetUserInfo() (*UserInfoWithMenus, error) {
Status: rpcResp.Status,
RoleId: rpcResp.RoleId,
RoleName: rpcResp.RoleName,
TenantId: tenantId,
LastLoginTime: rpcResp.LastLoginTime,
CreatedAt: rpcResp.CreatedAt,
UpdatedAt: rpcResp.UpdatedAt,

13
gateway/internal/logic/auth/loginLogic.go
View File

@ -2,6 +2,7 @@ package auth
import (
"context"
"errors"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
@ -34,11 +35,22 @@ func (l *LoginLogic) Login(req *types.LoginReq) (resp *types.LoginResp, err erro
return nil, err
}
if req.TenantId == "" {
return nil, errors.New("请选择要登录的租户")
}
verified, verr := l.svcCtx.CrmRepo.VerifyUserTenant(l.ctx, rpcResp.UserId, req.TenantId)
if verr != nil || !verified {
return nil, errors.New("用户不属于该租户，禁止登录")
}
tenantId := req.TenantId
token, err := jwtpkg.GenerateToken(
l.svcCtx.Config.Auth.AccessSecret,
l.svcCtx.Config.Auth.AccessExpire,
rpcResp.UserId,
rpcResp.RoleKey,
tenantId,
)
if err != nil {
return nil, err
@ -52,6 +64,7 @@ func (l *LoginLogic) Login(req *types.LoginReq) (resp *types.LoginResp, err erro
RealName: rpcResp.RealName,
Avatar: rpcResp.Avatar,
RoleName: rpcResp.RoleName,
TenantId: tenantId,
},
}, nil
}

32
gateway/internal/logic/crm/relation/common.go Normal file
View File

@ -0,0 +1,32 @@
package relation
import (
"database/sql"
"strings"
"time"
)
const timeLayout = "2006-01-02 15:04:05"
func parseTimeOrNow(raw string) time.Time {
if strings.TrimSpace(raw) == "" {
return time.Now()
}
t, err := time.ParseInLocation(timeLayout, raw, time.Local)
if err != nil {
return time.Now()
}
return t
}
func parseNullTime(raw string) sql.NullTime {
if strings.TrimSpace(raw) == "" {
return sql.NullTime{}
}
t, err := time.ParseInLocation(timeLayout, raw, time.Local)
if err != nil {
return sql.NullTime{}
}
return sql.NullTime{Time: t, Valid: true}
}

55
gateway/internal/logic/crm/relation/createRelationLogic.go Normal file
View File

@ -0,0 +1,55 @@
package relation
import (
"context"
"errors"
"muyu-apiserver/gateway/internal/repo"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"muyu-apiserver/pkg/ctxdata"
"github.com/zeromicro/go-zero/core/logx"
)
type CreateRelationLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewCreateRelationLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateRelationLogic {
return &CreateRelationLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *CreateRelationLogic) CreateRelation(req *types.CrmCreateRelationReq) (*types.IdResp, error) {
if req.FromTenantId == "" || req.ToTenantId == "" || req.RelationType == "" {
return nil, errors.New("fromTenantId/toTenantId/relationType are required")
}
if req.FromTenantId == req.ToTenantId {
return nil, errors.New("fromTenantId and toTenantId cannot be equal")
}
tenantId := ctxdata.GetTenantId(l.ctx)
userId := ctxdata.GetUserId(l.ctx)
relID, err := l.svcCtx.CrmRepo.CreateRelation(l.ctx, repo.CreateRelationInput{
OwnerTenantId: tenantId,
FromTenantId: req.FromTenantId,
ToTenantId: req.ToTenantId,
RelationType: req.RelationType,
Status: 1,
ValidFrom: parseTimeOrNow(req.ValidFrom),
ValidTo: parseNullTime(req.ValidTo),
Source: "manual",
CreatedBy: userId,
})
if err != nil {
return nil, err
}
return &types.IdResp{Id: relID}, nil
}

58
gateway/internal/logic/crm/relation/listDownstreamLogic.go Normal file
View File

@ -0,0 +1,58 @@
package relation
import (
"context"
"errors"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"muyu-apiserver/pkg/ctxdata"
"github.com/zeromicro/go-zero/core/logx"
)
type ListDownstreamLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewListDownstreamLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ListDownstreamLogic {
return &ListDownstreamLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *ListDownstreamLogic) ListDownstream(req *types.CrmListRelationsReq) (*types.CrmListRelationsResp, error) {
if req.Depth != 1 {
return nil, errors.New("only depth=1 is allowed")
}
tenantId := ctxdata.GetTenantId(l.ctx)
rows, err := l.svcCtx.CrmRepo.ListDownstream(l.ctx, tenantId)
if err != nil {
return nil, err
}
list := make([]types.CrmRelationInfo, 0, len(rows))
for _, row := range rows {
item := types.CrmRelationInfo{
RelationshipId: row.RelationshipId,
FromTenantId: row.FromTenantId,
FromTenantName: row.FromTenantName,
ToTenantId: row.ToTenantId,
ToTenantName: row.ToTenantName,
RelationType: row.RelationType,
Status: row.Status,
ValidFrom: row.ValidFrom.Format(timeLayout),
}
if row.ValidTo.Valid {
item.ValidTo = row.ValidTo.Time.Format(timeLayout)
}
list = append(list, item)
}
return &types.CrmListRelationsResp{Total: int64(len(list)), List: list}, nil
}

56
gateway/internal/logic/crm/relation/listRelationHistoryLogic.go Normal file
View File

@ -0,0 +1,56 @@
package relation
import (
"context"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"muyu-apiserver/pkg/ctxdata"
"github.com/zeromicro/go-zero/core/logx"
)
type ListRelationHistoryLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewListRelationHistoryLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ListRelationHistoryLogic {
return &ListRelationHistoryLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *ListRelationHistoryLogic) ListRelationHistory(req *types.CrmRelationHistoryReq) (*types.CrmRelationHistoryResp, error) {
tenantId := ctxdata.GetTenantId(l.ctx)
rows, total, err := l.svcCtx.CrmRepo.ListHistory(l.ctx, tenantId, req.Page, req.PageSize)
if err != nil {
return nil, err
}
list := make([]types.CrmRelationHistoryItem, 0, len(rows))
for _, row := range rows {
before := ""
if row.BeforeJSON.Valid {
before = row.BeforeJSON.String
}
after := ""
if row.AfterJSON.Valid {
after = row.AfterJSON.String
}
list = append(list, types.CrmRelationHistoryItem{
RelationshipId: row.RelationshipId,
ChangeType: row.ChangeType,
BeforeJson: before,
AfterJson: after,
ChangedBy: row.ChangedBy,
ChangedAt: row.ChangedAt.Format(timeLayout),
})
}
return &types.CrmRelationHistoryResp{Total: total, List: list}, nil
}

58
gateway/internal/logic/crm/relation/listUpstreamLogic.go Normal file
View File

@ -0,0 +1,58 @@
package relation
import (
"context"
"errors"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"muyu-apiserver/pkg/ctxdata"
"github.com/zeromicro/go-zero/core/logx"
)
type ListUpstreamLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewListUpstreamLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ListUpstreamLogic {
return &ListUpstreamLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *ListUpstreamLogic) ListUpstream(req *types.CrmListRelationsReq) (*types.CrmListRelationsResp, error) {
if req.Depth != 1 {
return nil, errors.New("only depth=1 is allowed")
}
tenantId := ctxdata.GetTenantId(l.ctx)
rows, err := l.svcCtx.CrmRepo.ListUpstream(l.ctx, tenantId)
if err != nil {
return nil, err
}
list := make([]types.CrmRelationInfo, 0, len(rows))
for _, row := range rows {
item := types.CrmRelationInfo{
RelationshipId: row.RelationshipId,
FromTenantId: row.FromTenantId,
FromTenantName: row.FromTenantName,
ToTenantId: row.ToTenantId,
ToTenantName: row.ToTenantName,
RelationType: row.RelationType,
Status: row.Status,
ValidFrom: row.ValidFrom.Format(timeLayout),
}
if row.ValidTo.Valid {
item.ValidTo = row.ValidTo.Time.Format(timeLayout)
}
list = append(list, item)
}
return &types.CrmListRelationsResp{Total: int64(len(list)), List: list}, nil
}

50
gateway/internal/logic/crm/relation/updateRelationLogic.go Normal file
View File

@ -0,0 +1,50 @@
package relation
import (
"context"
"errors"
"muyu-apiserver/gateway/internal/repo"
"muyu-apiserver/gateway/internal/svc"
"muyu-apiserver/gateway/internal/types"
"muyu-apiserver/pkg/ctxdata"
"github.com/zeromicro/go-zero/core/logx"
)
type UpdateRelationLogic struct {
logx.Logger
ctx context.Context
svcCtx *svc.ServiceContext
}
func NewUpdateRelationLogic(ctx context.Context, svcCtx *svc.ServiceContext) *UpdateRelationLogic {
return &UpdateRelationLogic{
Logger: logx.WithContext(ctx),
ctx: ctx,
svcCtx: svcCtx,
}
}
func (l *UpdateRelationLogic) UpdateRelation(req *types.CrmUpdateRelationReq) (*types.IdResp, error) {
relID := ctxdata.GetPathId(l.ctx)
if relID == "" {
return nil, errors.New("relationship id is required")
}
tenantId := ctxdata.GetTenantId(l.ctx)
userId := ctxdata.GetUserId(l.ctx)
if err := l.svcCtx.CrmRepo.UpdateRelation(l.ctx, repo.UpdateRelationInput{
RelationshipId: relID,
OwnerTenantId: tenantId,
Status: req.Status,
ValidFrom: parseTimeOrNow(req.ValidFrom),
ValidTo: parseNullTime(req.ValidTo),
UpdatedBy: userId,
}); err != nil {
return nil, err
}
return &types.IdResp{Id: relID}, nil
}

53
gateway/internal/middleware/authorityMiddleware.go
View File

@ -1,19 +1,62 @@
package middleware
import (
"encoding/json"
"net/http"
"github.com/casbin/casbin/v2"
pkgcasbin "muyu-apiserver/pkg/casbin"
"muyu-apiserver/pkg/ctxdata"
)
type AuthorityMiddleware struct{}
type AuthorityMiddleware struct {
enforcer *casbin.Enforcer
}
func NewAuthorityMiddleware() *AuthorityMiddleware {
return &AuthorityMiddleware{}
func NewAuthorityMiddleware(enforcer *casbin.Enforcer) *AuthorityMiddleware {
return &AuthorityMiddleware{enforcer: enforcer}
}
func (m *AuthorityMiddleware) Handle(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
// go-zero JWT middleware puts claims into context
// Casbin enforcement will be added later
// JWT claims are injected by go-zero middleware into context.
// Keep a hard guard here to prevent tenantless requests from passing through.
userId := ctxdata.GetUserId(r.Context())
roleKey := ctxdata.GetRoleKey(r.Context())
tenantId := ctxdata.GetTenantId(r.Context())
if userId == "" || roleKey == "" || tenantId == "" {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusUnauthorized)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 1004,
"msg": "invalid auth claims: tenant scope required",
})
return
}
// Tenant scope cannot be overridden by caller.
if hdrTenant := r.Header.Get("X-Tenant-Id"); hdrTenant != "" && hdrTenant != tenantId {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusForbidden)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 1006,
"msg": "tenant scope mismatch",
})
return
}
r.Header.Set("X-Tenant-Id", tenantId)
// Keep permissions closed by default once enforcer is enabled.
if m.enforcer != nil && !pkgcasbin.CheckPermission(m.enforcer, roleKey, r.URL.Path, r.Method) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusForbidden)
_ = json.NewEncoder(w).Encode(map[string]interface{}{
"code": 1006,
"msg": "access denied",
})
return
}
next(w, r)
}
}

250
gateway/internal/repo/crmrepo.go Normal file
View File

@ -0,0 +1,250 @@
package repo
import (
"context"
"database/sql"
"errors"
"time"
"muyu-apiserver/pkg/uid"
"github.com/zeromicro/go-zero/core/stores/sqlx"
)
type (
CrmRepo struct {
conn sqlx.SqlConn
}
RelationView struct {
RelationshipId string `db:"relationship_id" json:"relationshipId"`
FromTenantId string `db:"from_tenant_id" json:"fromTenantId"`
FromTenantName string `db:"from_tenant_name" json:"fromTenantName"`
ToTenantId string `db:"to_tenant_id" json:"toTenantId"`
ToTenantName string `db:"to_tenant_name" json:"toTenantName"`
RelationType string `db:"relation_type" json:"relationType"`
Status int64 `db:"status" json:"status"`
ValidFrom time.Time `db:"valid_from" json:"validFrom"`
ValidTo sql.NullTime `db:"valid_to" json:"validTo,omitempty"`
}
CreateRelationInput struct {
OwnerTenantId string
FromTenantId string
ToTenantId string
RelationType string
Status int64
ValidFrom time.Time
ValidTo sql.NullTime
Source string
CreatedBy string
}
UpdateRelationInput struct {
RelationshipId string
OwnerTenantId string
Status int64
ValidFrom time.Time
ValidTo sql.NullTime
UpdatedBy string
}
RelationHistory struct {
RelationshipId string `db:"relationship_id" json:"relationshipId"`
ChangeType string `db:"change_type" json:"changeType"`
BeforeJSON sql.NullString `db:"before_json" json:"beforeJson"`
AfterJSON sql.NullString `db:"after_json" json:"afterJson"`
ChangedBy string `db:"changed_by" json:"changedBy"`
ChangedAt time.Time `db:"changed_at" json:"changedAt"`
}
)
func NewCrmRepo(conn sqlx.SqlConn) *CrmRepo {
return &CrmRepo{conn: conn}
}
func (r *CrmRepo) VerifyUserTenant(ctx context.Context, userId, tenantId string) (bool, error) {
var count int64
err := r.conn.QueryRowCtx(ctx, &count,
"SELECT COUNT(1) FROM crm_tenant_user WHERE user_id = $1 AND tenant_id = $2 AND status = 1",
userId, tenantId,
)
if err != nil {
return false, err
}
return count > 0, nil
}
func (r *CrmRepo) ResolvePrimaryTenant(ctx context.Context, userId string) (string, error) {
var tenantId string
err := r.conn.QueryRowCtx(ctx, &tenantId,
"SELECT tenant_id FROM crm_tenant_user WHERE user_id = $1 AND status = 1 ORDER BY is_primary DESC, id ASC LIMIT 1",
userId,
)
if err != nil {
return "", err
}
return tenantId, nil
}
func (r *CrmRepo) ListUpstream(ctx context.Context, tenantId string) ([]RelationView, error) {
rows := make([]RelationView, 0)
query := `
SELECT r.relationship_id, r.from_tenant_id, ft.tenant_name AS from_tenant_name,
r.to_tenant_id, tt.tenant_name AS to_tenant_name,
r.relation_type, r.status, r.valid_from, r.valid_to
FROM crm_tenant_relationship r
JOIN crm_tenant ft ON ft.tenant_id = r.from_tenant_id
JOIN crm_tenant tt ON tt.tenant_id = r.to_tenant_id
WHERE r.owner_tenant_id = $1
AND r.to_tenant_id = $2
AND r.status = 1
AND r.valid_from <= NOW()
AND (r.valid_to IS NULL OR r.valid_to > NOW())
ORDER BY r.updated_at DESC`
if err := r.conn.QueryRowsCtx(ctx, &rows, query, tenantId, tenantId); err != nil {
return nil, err
}
return rows, nil
}
func (r *CrmRepo) ListDownstream(ctx context.Context, tenantId string) ([]RelationView, error) {
rows := make([]RelationView, 0)
query := `
SELECT r.relationship_id, r.from_tenant_id, ft.tenant_name AS from_tenant_name,
r.to_tenant_id, tt.tenant_name AS to_tenant_name,
r.relation_type, r.status, r.valid_from, r.valid_to
FROM crm_tenant_relationship r
JOIN crm_tenant ft ON ft.tenant_id = r.from_tenant_id
JOIN crm_tenant tt ON tt.tenant_id = r.to_tenant_id
WHERE r.owner_tenant_id = $1
AND r.from_tenant_id = $2
AND r.status = 1
AND r.valid_from <= NOW()
AND (r.valid_to IS NULL OR r.valid_to > NOW())
ORDER BY r.updated_at DESC`
if err := r.conn.QueryRowsCtx(ctx, &rows, query, tenantId, tenantId); err != nil {
return nil, err
}
return rows, nil
}
func (r *CrmRepo) CreateRelation(ctx context.Context, in CreateRelationInput) (string, error) {
relID := "rel_" + uid.Generate()[:16]
if in.Source == "" {
in.Source = "manual"
}
if in.Status == 0 {
in.Status = 1
}
_, err := r.conn.ExecCtx(ctx, `
INSERT INTO crm_tenant_relationship
(relationship_id, owner_tenant_id, from_tenant_id, to_tenant_id, relation_type, status, valid_from, valid_to, source, created_by)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`,
relID, in.OwnerTenantId, in.FromTenantId, in.ToTenantId, in.RelationType,
in.Status, in.ValidFrom, in.ValidTo, in.Source, in.CreatedBy)
if err != nil {
return "", err
}
histID := "rh_" + uid.Generate()[:16]
_, _ = r.conn.ExecCtx(ctx, `
INSERT INTO crm_tenant_relationship_history
(history_id, relationship_id, change_type, before_json, after_json, changed_by)
VALUES ($1, $2, 'create', NULL, jsonb_build_object('status', $3::text, 'relationType', $4), $5)`,
histID, relID, in.Status, in.RelationType, in.CreatedBy)
return relID, nil
}
func (r *CrmRepo) UpdateRelation(ctx context.Context, in UpdateRelationInput) error {
res, err := r.conn.ExecCtx(ctx, `
UPDATE crm_tenant_relationship
SET status = $1, valid_from = $2, valid_to = $3, updated_at = NOW()
WHERE relationship_id = $4 AND owner_tenant_id = $5`,
in.Status, in.ValidFrom, in.ValidTo, in.RelationshipId, in.OwnerTenantId)
if err != nil {
return err
}
affected, _ := res.RowsAffected()
if affected == 0 {
return errors.New("relationship not found")
}
histID := "rh_" + uid.Generate()[:16]
_, _ = r.conn.ExecCtx(ctx, `
INSERT INTO crm_tenant_relationship_history
(history_id, relationship_id, change_type, before_json, after_json, changed_by)
VALUES ($1, $2, 'update', NULL, jsonb_build_object('status', $3::text), $4)`,
histID, in.RelationshipId, in.Status, in.UpdatedBy)
return nil
}
func (r *CrmRepo) ListHistory(ctx context.Context, ownerTenantId string, page, pageSize int64) ([]RelationHistory, int64, error) {
if page < 1 {
page = 1
}
if pageSize < 1 {
pageSize = 20
}
var total int64
if err := r.conn.QueryRowCtx(ctx, &total, `
SELECT COUNT(1)
FROM crm_tenant_relationship_history h
JOIN crm_tenant_relationship r ON r.relationship_id = h.relationship_id
WHERE r.owner_tenant_id = $1`, ownerTenantId); err != nil {
return nil, 0, err
}
offset := (page - 1) * pageSize
rows := make([]RelationHistory, 0)
if err := r.conn.QueryRowsCtx(ctx, &rows, `
SELECT h.relationship_id, h.change_type, h.before_json, h.after_json, h.changed_by, h.changed_at
FROM crm_tenant_relationship_history h
JOIN crm_tenant_relationship r ON r.relationship_id = h.relationship_id
WHERE r.owner_tenant_id = $1
ORDER BY h.changed_at DESC
LIMIT $2 OFFSET $3`, ownerTenantId, pageSize, offset); err != nil {
return nil, 0, err
}
return rows, total, nil
}
type GraphNode struct {
TenantId string `db:"tenant_id" json:"id"`
TenantName string `db:"tenant_name" json:"name"`
}
type GraphEdge struct {
Source string `db:"from_tenant_id" json:"source"`
Target string `db:"to_tenant_id" json:"target"`
RelationType string `db:"relation_type" json:"relationType"`
RelationshipId string `db:"relationship_id" json:"relationshipId"`
}
type FullGraph struct {
Nodes []GraphNode `json:"nodes"`
Edges []GraphEdge `json:"edges"`
}
func (r *CrmRepo) GetFullGraph(ctx context.Context) (*FullGraph, error) {
nodes := make([]GraphNode, 0)
if err := r.conn.QueryRowsCtx(ctx, &nodes,
"SELECT tenant_id, tenant_name FROM crm_tenant WHERE status = 1 ORDER BY tenant_id"); err != nil {
return nil, err
}
edges := make([]GraphEdge, 0)
if err := r.conn.QueryRowsCtx(ctx, &edges, `
SELECT DISTINCT r.relationship_id, r.from_tenant_id, r.to_tenant_id, r.relation_type
FROM crm_tenant_relationship r
WHERE r.status = 1
AND r.valid_from <= NOW()
AND (r.valid_to IS NULL OR r.valid_to > NOW())
ORDER BY r.from_tenant_id, r.to_tenant_id`); err != nil {
return nil, err
}
return &FullGraph{Nodes: nodes, Edges: edges}, nil
}

43
gateway/internal/svc/serviceContext.go
View File

@ -1,10 +1,17 @@
package svc
import (
"github.com/casbin/casbin/v2"
_ "github.com/lib/pq"
"github.com/zeromicro/go-zero/core/logx"
"github.com/zeromicro/go-zero/core/stores/sqlx"
"github.com/zeromicro/go-zero/rest"
"github.com/zeromicro/go-zero/zrpc"
"muyu-apiserver/gateway/internal/config"
"muyu-apiserver/gateway/internal/middleware"
"muyu-apiserver/gateway/internal/repo"
pkgcasbin "muyu-apiserver/pkg/casbin"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/inventory/inventoryservice"
"muyu-apiserver/rpc/system/systemservice"
)
@ -12,15 +19,45 @@ import (
type ServiceContext struct {
Config config.Config
AuthorityMiddleware rest.Middleware
CasbinEnforcer *casbin.Enforcer
CrmRepo *repo.CrmRepo
SystemRpc systemservice.SystemService
InventoryRpc inventoryservice.InventoryService
}
func NewServiceContext(c config.Config) *ServiceContext {
mysqlConn := sqlx.NewMysql(c.DataSource)
var pgConn sqlx.SqlConn
if c.PostgresDataSource != "" {
pgConn = sqlx.NewSqlConn("postgres", c.PostgresDataSource)
} else {
pgConn = mysqlConn
}
var enforcer *casbin.Enforcer
if c.DataSource != "" {
func() {
defer func() {
if r := recover(); r != nil {
logx.Errorf("init casbin enforcer panic: %v (skipped, running without enforcer)", r)
}
}()
e, err := pkgcasbin.NewEnforcer(c.DataSource)
if err != nil {
logx.Errorf("init casbin enforcer failed: %v", err)
return
}
enforcer = e
}()
}
return &ServiceContext{
Config: c,
AuthorityMiddleware: middleware.NewAuthorityMiddleware().Handle,
SystemRpc: systemservice.NewSystemService(zrpc.MustNewClient(c.SystemRpc)),
InventoryRpc: inventoryservice.NewInventoryService(zrpc.MustNewClient(c.InventoryRpc)),
AuthorityMiddleware: middleware.NewAuthorityMiddleware(enforcer).Handle,
CasbinEnforcer: enforcer,
CrmRepo: repo.NewCrmRepo(pgConn),
SystemRpc: systemservice.NewSystemService(zrpc.MustNewClient(c.SystemRpc, zrpc.WithUnaryClientInterceptor(tenantctx.UnaryClientInterceptor()))),
InventoryRpc: inventoryservice.NewInventoryService(zrpc.MustNewClient(c.InventoryRpc, zrpc.WithUnaryClientInterceptor(tenantctx.UnaryClientInterceptor()))),
}
}

56
gateway/internal/types/types.go
View File

@ -185,6 +185,7 @@ type ListUserResp struct {
type LoginReq struct {
Username string `json:"username"`
Password string `json:"password"`
TenantId string `json:"tenantId,optional"`
}
type LoginResp struct {
@ -405,7 +406,62 @@ type UserInfo struct {
Status int64 `json:"status"`
RoleId string `json:"roleId"`
RoleName string `json:"roleName"`
TenantId string `json:"tenantId"`
LastLoginTime string `json:"lastLoginTime"`
CreatedAt string `json:"createdAt"`
UpdatedAt string `json:"updatedAt"`
}
type CrmRelationInfo struct {
RelationshipId string `json:"relationshipId"`
FromTenantId string `json:"fromTenantId"`
FromTenantName string `json:"fromTenantName"`
ToTenantId string `json:"toTenantId"`
ToTenantName string `json:"toTenantName"`
RelationType string `json:"relationType"`
Status int64 `json:"status"`
ValidFrom string `json:"validFrom"`
ValidTo string `json:"validTo,optional"`
}
type CrmListRelationsReq struct {
Depth int64 `form:"depth,default=1"`
}
type CrmListRelationsResp struct {
Total int64 `json:"total"`
List []CrmRelationInfo `json:"list"`
}
type CrmCreateRelationReq struct {
FromTenantId string `json:"fromTenantId"`
ToTenantId string `json:"toTenantId"`
RelationType string `json:"relationType"`
ValidFrom string `json:"validFrom,optional"`
ValidTo string `json:"validTo,optional"`
}
type CrmUpdateRelationReq struct {
Status int64 `json:"status"`
ValidFrom string `json:"validFrom,optional"`
ValidTo string `json:"validTo,optional"`
}
type CrmRelationHistoryReq struct {
Page int64 `form:"page,default=1"`
PageSize int64 `form:"pageSize,default=20"`
}
type CrmRelationHistoryItem struct {
RelationshipId string `json:"relationshipId"`
ChangeType string `json:"changeType"`
BeforeJson string `json:"beforeJson"`
AfterJson string `json:"afterJson"`
ChangedBy string `json:"changedBy"`
ChangedAt string `json:"changedAt"`
}
type CrmRelationHistoryResp struct {
Total int64 `json:"total"`
List []CrmRelationHistoryItem `json:"list"`
}

5
go.mod
View File

@ -4,8 +4,9 @@ go 1.24.4
require (
github.com/casbin/casbin/v2 v2.135.0
github.com/casbin/gorm-adapter/v3 v3.41.0
github.com/casbin/gorm-adapter/v3 v3.28.0
github.com/golang-jwt/jwt/v5 v5.3.1
github.com/lib/pq v1.12.0
github.com/zeromicro/go-zero v1.10.0
golang.org/x/crypto v0.48.0
google.golang.org/grpc v1.79.1
@ -18,7 +19,6 @@ require (
filippo.io/edwards25519 v1.1.0 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/bmatcuk/doublestar/v4 v4.9.1 // indirect
github.com/casbin/casbin/v3 v3.8.1 // indirect
github.com/casbin/govaluate v1.10.0 // indirect
github.com/cenkalti/backoff/v4 v4.3.0 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
@ -68,7 +68,6 @@ require (
github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/openzipkin/zipkin-go v0.4.3 // indirect
github.com/pelletier/go-toml/v2 v2.2.4 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/prometheus/client_golang v1.23.2 // indirect
github.com/prometheus/client_model v0.6.2 // indirect
github.com/prometheus/common v0.66.1 // indirect

10
go.sum
View File

@ -41,10 +41,8 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
github.com/casbin/casbin/v2 v2.135.0 h1:6BLkMQiGotYyS5yYeWgW19vxqugUlvHFkFiLnLR/bxk=
github.com/casbin/casbin/v2 v2.135.0/go.mod h1:FmcfntdXLTcYXv/hxgNntcRPqAbwOG9xsism0yXT+18=
github.com/casbin/casbin/v3 v3.8.1 h1:D4dEY4knePPR4YgNP5WZtWNaOxD0UK0LpPy9+zxtBwo=
github.com/casbin/casbin/v3 v3.8.1/go.mod h1:5rJbQr2e6AuuDDNxnPc5lQlC9nIgg6nS1zYwKXhpHC8=
github.com/casbin/gorm-adapter/v3 v3.41.0 h1:Xhpi0tfRP9aKPDWDf6dgBxHZ9UM6IophxxPIEGWqCNM=
github.com/casbin/gorm-adapter/v3 v3.41.0/go.mod h1:BQZRJhwUnwMpI+pT2m7/cUJwXxrHfzpBpPcNTyMGeGA=
github.com/casbin/gorm-adapter/v3 v3.28.0 h1:ORF8prF6SfaipdgT1fud+r1Tp5J0uul8QaKJHqCPY/o=
github.com/casbin/gorm-adapter/v3 v3.28.0/go.mod h1:aftWi0cla0CC1bHQVrSFzBcX/98IFK28AvuPppCQgTs=
github.com/casbin/govaluate v1.3.0/go.mod h1:G/UnbIjZk/0uMNaLwZZmFQrR72tYRZWQkO70si/iR7A=
github.com/casbin/govaluate v1.10.0 h1:ffGw51/hYH3w3rZcxO/KcaUIDOLP84w7nsidMVgaDG0=
github.com/casbin/govaluate v1.10.0/go.mod h1:G/UnbIjZk/0uMNaLwZZmFQrR72tYRZWQkO70si/iR7A=
@ -172,8 +170,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/lib/pq v1.10.2 h1:AqzbZs4ZoCBp+GtejcpCpcxM3zlSMx29dXbUSeVtJb8=
github.com/lib/pq v1.10.2/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
github.com/lib/pq v1.12.0 h1:mC1zeiNamwKBecjHarAr26c/+d8V5w/u4J0I/yASbJo=
github.com/lib/pq v1.12.0/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA=
github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=

32
model/invproductmodel.go
View File

@ -19,10 +19,10 @@ type StockGroupResult struct {
type (
InvProductModel interface {
invProductModel
FindList(ctx context.Context, page, pageSize int64, productName, spec, color, location string, status int64) ([]*InvProduct, int64, error)
FindStockSummary(ctx context.Context) (productCount, totalPieces, totalRolls int64, totalCostValue, totalSalesValue float64, err error)
FindGroupByColor(ctx context.Context) ([]StockGroupResult, error)
FindGroupByLocation(ctx context.Context) ([]StockGroupResult, error)
FindList(ctx context.Context, tenantId string, page, pageSize int64, productName, spec, color, location string, status int64) ([]*InvProduct, int64, error)
FindStockSummary(ctx context.Context, tenantId string) (productCount, totalPieces, totalRolls int64, totalCostValue, totalSalesValue float64, err error)
FindGroupByColor(ctx context.Context, tenantId string) ([]StockGroupResult, error)
FindGroupByLocation(ctx context.Context, tenantId string) ([]StockGroupResult, error)
}
customInvProductModel struct {
@ -36,9 +36,9 @@ func NewInvProductModel(conn sqlx.SqlConn, c cache.CacheConf, opts ...cache.Opti
}
}
func (m *customInvProductModel) FindList(ctx context.Context, page, pageSize int64, productName, spec, color, location string, status int64) ([]*InvProduct, int64, error) {
where := "WHERE deleted_at IS NULL"
args := make([]interface{}, 0)
func (m *customInvProductModel) FindList(ctx context.Context, tenantId string, page, pageSize int64, productName, spec, color, location string, status int64) ([]*InvProduct, int64, error) {
where := "WHERE deleted_at IS NULL AND tenant_id = ?"
args := []interface{}{tenantId}
if productName != "" {
where += " AND product_name LIKE ?"
@ -79,7 +79,7 @@ func (m *customInvProductModel) FindList(ctx context.Context, page, pageSize int
return list, total, nil
}
func (m *customInvProductModel) FindStockSummary(ctx context.Context) (productCount, totalPieces, totalRolls int64, totalCostValue, totalSalesValue float64, err error) {
func (m *customInvProductModel) FindStockSummary(ctx context.Context, tenantId string) (productCount, totalPieces, totalRolls int64, totalCostValue, totalSalesValue float64, err error) {
var summary struct {
ProductCount int64 `db:"product_count"`
TotalPieces int64 `db:"total_pieces"`
@ -87,28 +87,28 @@ func (m *customInvProductModel) FindStockSummary(ctx context.Context) (productCo
TotalCostValue float64 `db:"total_cost_value"`
TotalSalesValue float64 `db:"total_sales_value"`
}
query := fmt.Sprintf("SELECT COUNT(*) AS product_count, COALESCE(SUM(unit_pieces), 0) AS total_pieces, COALESCE(SUM(unit_rolls), 0) AS total_rolls, COALESCE(SUM(stock_quantity * cost_price), 0) AS total_cost_value, COALESCE(SUM(stock_quantity * sales_price), 0) AS total_sales_value FROM %s WHERE deleted_at IS NULL", m.table)
err = m.QueryRowNoCacheCtx(ctx, &summary, query)
query := fmt.Sprintf("SELECT COUNT(*) AS product_count, COALESCE(SUM(unit_pieces), 0) AS total_pieces, COALESCE(SUM(unit_rolls), 0) AS total_rolls, COALESCE(SUM(stock_quantity * cost_price), 0) AS total_cost_value, COALESCE(SUM(stock_quantity * sales_price), 0) AS total_sales_value FROM %s WHERE deleted_at IS NULL AND tenant_id = ?", m.table)
err = m.QueryRowNoCacheCtx(ctx, &summary, query, tenantId)
if err != nil {
return 0, 0, 0, 0, 0, err
}
return summary.ProductCount, summary.TotalPieces, summary.TotalRolls, summary.TotalCostValue, summary.TotalSalesValue, nil
}
func (m *customInvProductModel) FindGroupByColor(ctx context.Context) ([]StockGroupResult, error) {
func (m *customInvProductModel) FindGroupByColor(ctx context.Context, tenantId string) ([]StockGroupResult, error) {
var list []StockGroupResult
query := fmt.Sprintf("SELECT color AS name, COUNT(*) AS count, COALESCE(SUM(stock_quantity), 0) AS quantity FROM %s WHERE deleted_at IS NULL GROUP BY color ORDER BY count DESC", m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query)
query := fmt.Sprintf("SELECT color AS name, COUNT(*) AS count, COALESCE(SUM(stock_quantity), 0) AS quantity FROM %s WHERE deleted_at IS NULL AND tenant_id = ? GROUP BY color ORDER BY count DESC", m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query, tenantId)
if err != nil {
return nil, err
}
return list, nil
}
func (m *customInvProductModel) FindGroupByLocation(ctx context.Context) ([]StockGroupResult, error) {
func (m *customInvProductModel) FindGroupByLocation(ctx context.Context, tenantId string) ([]StockGroupResult, error) {
var list []StockGroupResult
query := fmt.Sprintf("SELECT location AS name, COUNT(*) AS count, COALESCE(SUM(stock_quantity), 0) AS quantity FROM %s WHERE deleted_at IS NULL GROUP BY location ORDER BY count DESC", m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query)
query := fmt.Sprintf("SELECT location AS name, COUNT(*) AS count, COALESCE(SUM(stock_quantity), 0) AS quantity FROM %s WHERE deleted_at IS NULL AND tenant_id = ? GROUP BY location ORDER BY count DESC", m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query, tenantId)
if err != nil {
return nil, err
}

8
model/invstockadjustmodel.go
View File

@ -13,7 +13,7 @@ var _ InvStockAdjustModel = (*customInvStockAdjustModel)(nil)
type (
InvStockAdjustModel interface {
invStockAdjustModel
FindList(ctx context.Context, page, pageSize int64, adjustNo, adjustReason string, status int64, startDate, endDate string) ([]*InvStockAdjust, int64, error)
FindList(ctx context.Context, tenantId string, page, pageSize int64, adjustNo, adjustReason string, status int64, startDate, endDate string) ([]*InvStockAdjust, int64, error)
}
customInvStockAdjustModel struct {
@ -27,9 +27,9 @@ func NewInvStockAdjustModel(conn sqlx.SqlConn, c cache.CacheConf, opts ...cache.
}
}
func (m *customInvStockAdjustModel) FindList(ctx context.Context, page, pageSize int64, adjustNo, adjustReason string, status int64, startDate, endDate string) ([]*InvStockAdjust, int64, error) {
where := "WHERE 1=1"
args := make([]interface{}, 0)
func (m *customInvStockAdjustModel) FindList(ctx context.Context, tenantId string, page, pageSize int64, adjustNo, adjustReason string, status int64, startDate, endDate string) ([]*InvStockAdjust, int64, error) {
where := "WHERE tenant_id = ?"
args := []interface{}{tenantId}
if adjustNo != "" {
where += " AND adjust_no LIKE ?"

8
model/invstockcheckmodel.go
View File

@ -13,7 +13,7 @@ var _ InvStockCheckModel = (*customInvStockCheckModel)(nil)
type (
InvStockCheckModel interface {
invStockCheckModel
FindList(ctx context.Context, page, pageSize int64, checkNo string, status int64, startDate, endDate string) ([]*InvStockCheck, int64, error)
FindList(ctx context.Context, tenantId string, page, pageSize int64, checkNo string, status int64, startDate, endDate string) ([]*InvStockCheck, int64, error)
}
customInvStockCheckModel struct {
@ -27,9 +27,9 @@ func NewInvStockCheckModel(conn sqlx.SqlConn, c cache.CacheConf, opts ...cache.O
}
}
func (m *customInvStockCheckModel) FindList(ctx context.Context, page, pageSize int64, checkNo string, status int64, startDate, endDate string) ([]*InvStockCheck, int64, error) {
where := "WHERE 1=1"
args := make([]interface{}, 0)
func (m *customInvStockCheckModel) FindList(ctx context.Context, tenantId string, page, pageSize int64, checkNo string, status int64, startDate, endDate string) ([]*InvStockCheck, int64, error) {
where := "WHERE tenant_id = ?"
args := []interface{}{tenantId}
if checkNo != "" {
where += " AND check_no LIKE ?"

12
model/invstockimportlogmodel.go
View File

@ -13,7 +13,7 @@ var _ InvStockImportLogModel = (*customInvStockImportLogModel)(nil)
type (
InvStockImportLogModel interface {
invStockImportLogModel
FindList(ctx context.Context, page, pageSize int64) ([]*InvStockImportLog, int64, error)
FindList(ctx context.Context, tenantId string, page, pageSize int64) ([]*InvStockImportLog, int64, error)
}
customInvStockImportLogModel struct {
@ -27,17 +27,17 @@ func NewInvStockImportLogModel(conn sqlx.SqlConn, c cache.CacheConf, opts ...cac
}
}
func (m *customInvStockImportLogModel) FindList(ctx context.Context, page, pageSize int64) ([]*InvStockImportLog, int64, error) {
func (m *customInvStockImportLogModel) FindList(ctx context.Context, tenantId string, page, pageSize int64) ([]*InvStockImportLog, int64, error) {
var total int64
countQuery := fmt.Sprintf("SELECT COUNT(*) FROM %s", m.table)
err := m.QueryRowNoCacheCtx(ctx, &total, countQuery)
countQuery := fmt.Sprintf("SELECT COUNT(*) FROM %s WHERE tenant_id = ?", m.table)
err := m.QueryRowNoCacheCtx(ctx, &total, countQuery, tenantId)
if err != nil {
return nil, 0, err
}
var list []*InvStockImportLog
query := fmt.Sprintf("SELECT %s FROM %s ORDER BY created_at DESC LIMIT ? OFFSET ?", invStockImportLogRows, m.table)
err = m.QueryRowsNoCacheCtx(ctx, &list, query, pageSize, (page-1)*pageSize)
query := fmt.Sprintf("SELECT %s FROM %s WHERE tenant_id = ? ORDER BY created_at DESC LIMIT ? OFFSET ?", invStockImportLogRows, m.table)
err = m.QueryRowsNoCacheCtx(ctx, &list, query, tenantId, pageSize, (page-1)*pageSize)
if err != nil {
return nil, 0, err
}

24
model/sysconfigmodel.go
View File

@ -13,9 +13,9 @@ var _ SysConfigModel = (*customSysConfigModel)(nil)
type (
SysConfigModel interface {
sysConfigModel
FindByGroup(ctx context.Context, group string) ([]*SysConfig, error)
FindAll(ctx context.Context) ([]*SysConfig, error)
UpdateByKey(ctx context.Context, key, value string) error
FindByGroup(ctx context.Context, tenantId, group string) ([]*SysConfig, error)
FindAll(ctx context.Context, tenantId string) ([]*SysConfig, error)
UpdateByKey(ctx context.Context, tenantId, key, value string) error
}
customSysConfigModel struct {
@ -29,28 +29,28 @@ func NewSysConfigModel(conn sqlx.SqlConn, c cache.CacheConf, opts ...cache.Optio
}
}
func (m *customSysConfigModel) FindByGroup(ctx context.Context, group string) ([]*SysConfig, error) {
func (m *customSysConfigModel) FindByGroup(ctx context.Context, tenantId, group string) ([]*SysConfig, error) {
var list []*SysConfig
query := fmt.Sprintf("SELECT %s FROM %s WHERE `config_group` = ?", sysConfigRows, m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query, group)
query := fmt.Sprintf("SELECT %s FROM %s WHERE tenant_id = ? AND `config_group` = ?", sysConfigRows, m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query, tenantId, group)
if err != nil {
return nil, err
}
return list, nil
}
func (m *customSysConfigModel) FindAll(ctx context.Context) ([]*SysConfig, error) {
func (m *customSysConfigModel) FindAll(ctx context.Context, tenantId string) ([]*SysConfig, error) {
var list []*SysConfig
query := fmt.Sprintf("SELECT %s FROM %s", sysConfigRows, m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query)
query := fmt.Sprintf("SELECT %s FROM %s WHERE tenant_id = ?", sysConfigRows, m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query, tenantId)
if err != nil {
return nil, err
}
return list, nil
}
func (m *customSysConfigModel) UpdateByKey(ctx context.Context, key, value string) error {
query := fmt.Sprintf("UPDATE %s SET `config_value` = ? WHERE `config_key` = ?", m.table)
_, err := m.ExecNoCacheCtx(ctx, query, value, key)
func (m *customSysConfigModel) UpdateByKey(ctx context.Context, tenantId, key, value string) error {
query := fmt.Sprintf("UPDATE %s SET `config_value` = ? WHERE tenant_id = ? AND `config_key` = ?", m.table)
_, err := m.ExecNoCacheCtx(ctx, query, value, tenantId, key)
return err
}

16
model/sysrolemodel.go
View File

@ -13,8 +13,8 @@ var _ SysRoleModel = (*customSysRoleModel)(nil)
type (
SysRoleModel interface {
sysRoleModel
FindList(ctx context.Context, page, pageSize int64, roleName string, status int64) ([]*SysRole, int64, error)
FindAll(ctx context.Context) ([]*SysRole, error)
FindList(ctx context.Context, tenantId string, page, pageSize int64, roleName string, status int64) ([]*SysRole, int64, error)
FindAll(ctx context.Context, tenantId string) ([]*SysRole, error)
}
customSysRoleModel struct {
@ -28,9 +28,9 @@ func NewSysRoleModel(conn sqlx.SqlConn, c cache.CacheConf, opts ...cache.Option)
}
}
func (m *customSysRoleModel) FindList(ctx context.Context, page, pageSize int64, roleName string, status int64) ([]*SysRole, int64, error) {
where := "WHERE 1=1"
args := make([]interface{}, 0)
func (m *customSysRoleModel) FindList(ctx context.Context, tenantId string, page, pageSize int64, roleName string, status int64) ([]*SysRole, int64, error) {
where := "WHERE tenant_id = ?"
args := []interface{}{tenantId}
if roleName != "" {
where += " AND role_name LIKE ?"
@ -59,10 +59,10 @@ func (m *customSysRoleModel) FindList(ctx context.Context, page, pageSize int64,
return list, total, nil
}
func (m *customSysRoleModel) FindAll(ctx context.Context) ([]*SysRole, error) {
func (m *customSysRoleModel) FindAll(ctx context.Context, tenantId string) ([]*SysRole, error) {
var list []*SysRole
query := fmt.Sprintf("SELECT %s FROM %s ORDER BY sort_order ASC", sysRoleRows, m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query)
query := fmt.Sprintf("SELECT %s FROM %s WHERE tenant_id = ? ORDER BY sort_order ASC", sysRoleRows, m.table)
err := m.QueryRowsNoCacheCtx(ctx, &list, query, tenantId)
if err != nil {
return nil, err
}

8
model/sysusermodel.go
View File

@ -13,7 +13,7 @@ var _ SysUserModel = (*customSysUserModel)(nil)
type (
SysUserModel interface {
sysUserModel
FindList(ctx context.Context, page, pageSize int64, username, realName, phone string, status int64) ([]*SysUser, int64, error)
FindList(ctx context.Context, tenantId string, page, pageSize int64, username, realName, phone string, status int64) ([]*SysUser, int64, error)
}
customSysUserModel struct {
@ -27,9 +27,9 @@ func NewSysUserModel(conn sqlx.SqlConn, c cache.CacheConf, opts ...cache.Option)
}
}
func (m *customSysUserModel) FindList(ctx context.Context, page, pageSize int64, username, realName, phone string, status int64) ([]*SysUser, int64, error) {
where := "WHERE deleted_at IS NULL"
args := make([]interface{}, 0)
func (m *customSysUserModel) FindList(ctx context.Context, tenantId string, page, pageSize int64, username, realName, phone string, status int64) ([]*SysUser, int64, error) {
where := "WHERE deleted_at IS NULL AND tenant_id = ?"
args := []interface{}{tenantId}
if username != "" {
where += " AND username LIKE ?"

7
pkg/ctxdata/ctxdata.go
View File

@ -16,6 +16,13 @@ func GetRoleKey(ctx context.Context) string {
return ""
}
func GetTenantId(ctx context.Context) string {
if v, ok := ctx.Value("tenantId").(string); ok {
return v
}
return ""
}
func GetPathId(ctx context.Context) string {
if v, ok := ctx.Value("pathId").(string); ok {
return v

4
pkg/jwt/jwt.go
View File

@ -9,14 +9,16 @@ import (
type Claims struct {
UserId string `json:"userId"`
RoleKey string `json:"roleKey"`
TenantId string `json:"tenantId"`
jwt.RegisteredClaims
}
func GenerateToken(secret string, expire int64, userId, roleKey string) (string, error) {
func GenerateToken(secret string, expire int64, userId, roleKey, tenantId string) (string, error) {
now := time.Now()
claims := Claims{
UserId: userId,
RoleKey: roleKey,
TenantId: tenantId,
RegisteredClaims: jwt.RegisteredClaims{
ExpiresAt: jwt.NewNumericDate(now.Add(time.Duration(expire) * time.Second)),
IssuedAt: jwt.NewNumericDate(now),

35
pkg/tenantctx/tenantctx.go Normal file
View File

@ -0,0 +1,35 @@
package tenantctx
import (
"context"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
const headerKey = "x-tenant-id"
func InjectTenantId(ctx context.Context, tenantId string) context.Context {
return metadata.AppendToOutgoingContext(ctx, headerKey, tenantId)
}
func ExtractTenantId(ctx context.Context) string {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
return ""
}
vals := md.Get(headerKey)
if len(vals) == 0 {
return ""
}
return vals[0]
}
func UnaryClientInterceptor() grpc.UnaryClientInterceptor {
return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
if tid, ok := ctx.Value("tenantId").(string); ok && tid != "" {
ctx = metadata.AppendToOutgoingContext(ctx, headerKey, tid)
}
return invoker(ctx, method, req, reply, cc, opts...)
}
}

6
rpc/inventory/internal/logic/getstockgrouplogic.go
View File

@ -5,6 +5,7 @@ import (
"fmt"
"muyu-apiserver/model"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/inventory/internal/svc"
"muyu-apiserver/rpc/inventory/pb"
@ -28,14 +29,15 @@ func NewGetStockGroupLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Get
}
func (l *GetStockGroupLogic) GetStockGroup(in *pb.StockGroupReq) (*pb.StockGroupResp, error) {
tenantId := tenantctx.ExtractTenantId(l.ctx)
var list []model.StockGroupResult
var err error
switch in.GroupBy {
case "color":
list, err = l.svcCtx.ProductModel.FindGroupByColor(l.ctx)
list, err = l.svcCtx.ProductModel.FindGroupByColor(l.ctx, tenantId)
case "location":
list, err = l.svcCtx.ProductModel.FindGroupByLocation(l.ctx)
list, err = l.svcCtx.ProductModel.FindGroupByLocation(l.ctx, tenantId)
default:
return nil, status.Errorf(codes.InvalidArgument, "unsupported group_by: %s", in.GroupBy)
}

4
rpc/inventory/internal/logic/getstocksummarylogic.go
View File

@ -4,6 +4,7 @@ import (
"context"
"fmt"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/inventory/internal/svc"
"muyu-apiserver/rpc/inventory/pb"
@ -27,7 +28,8 @@ func NewGetStockSummaryLogic(ctx context.Context, svcCtx *svc.ServiceContext) *G
}
func (l *GetStockSummaryLogic) GetStockSummary(in *pb.StockSummaryReq) (*pb.StockSummaryResp, error) {
productCount, totalPieces, totalRolls, totalCostValue, totalSalesValue, err := l.svcCtx.ProductModel.FindStockSummary(l.ctx)
tenantId := tenantctx.ExtractTenantId(l.ctx)
productCount, totalPieces, totalRolls, totalCostValue, totalSalesValue, err := l.svcCtx.ProductModel.FindStockSummary(l.ctx, tenantId)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}

4
rpc/inventory/internal/logic/listimportloglogic.go
View File

@ -3,6 +3,7 @@ package logic
import (
"context"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/inventory/internal/svc"
"muyu-apiserver/rpc/inventory/pb"
@ -26,7 +27,8 @@ func NewListImportLogLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Lis
}
func (l *ListImportLogLogic) ListImportLog(in *pb.ListImportLogReq) (*pb.ListImportLogResp, error) {
list, total, err := l.svcCtx.ImportLogModel.FindList(l.ctx, in.Page, in.PageSize)
tenantId := tenantctx.ExtractTenantId(l.ctx)
list, total, err := l.svcCtx.ImportLogModel.FindList(l.ctx, tenantId, in.Page, in.PageSize)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}

4
rpc/inventory/internal/logic/listproductlogic.go
View File

@ -4,6 +4,7 @@ import (
"context"
"fmt"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/inventory/internal/svc"
"muyu-apiserver/rpc/inventory/pb"
@ -27,7 +28,8 @@ func NewListProductLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ListP
}
func (l *ListProductLogic) ListProduct(in *pb.ListProductReq) (*pb.ListProductResp, error) {
list, total, err := l.svcCtx.ProductModel.FindList(l.ctx, in.Page, in.PageSize, in.ProductName, in.Spec, in.Color, in.Location, in.Status)
tenantId := tenantctx.ExtractTenantId(l.ctx)
list, total, err := l.svcCtx.ProductModel.FindList(l.ctx, tenantId, in.Page, in.PageSize, in.ProductName, in.Spec, in.Color, in.Location, in.Status)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}

4
rpc/inventory/internal/logic/liststockadjustlogic.go
View File

@ -3,6 +3,7 @@ package logic
import (
"context"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/inventory/internal/svc"
"muyu-apiserver/rpc/inventory/pb"
@ -26,7 +27,8 @@ func NewListStockAdjustLogic(ctx context.Context, svcCtx *svc.ServiceContext) *L
}
func (l *ListStockAdjustLogic) ListStockAdjust(in *pb.ListStockAdjustReq) (*pb.ListStockAdjustResp, error) {
list, total, err := l.svcCtx.StockAdjustModel.FindList(l.ctx, in.Page, in.PageSize, in.AdjustNo, in.AdjustReason, in.Status, in.StartDate, in.EndDate)
tenantId := tenantctx.ExtractTenantId(l.ctx)
list, total, err := l.svcCtx.StockAdjustModel.FindList(l.ctx, tenantId, in.Page, in.PageSize, in.AdjustNo, in.AdjustReason, in.Status, in.StartDate, in.EndDate)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}

4
rpc/inventory/internal/logic/liststockchecklogic.go
View File

@ -3,6 +3,7 @@ package logic
import (
"context"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/inventory/internal/svc"
"muyu-apiserver/rpc/inventory/pb"
@ -26,7 +27,8 @@ func NewListStockCheckLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Li
}
func (l *ListStockCheckLogic) ListStockCheck(in *pb.ListStockCheckReq) (*pb.ListStockCheckResp, error) {
list, total, err := l.svcCtx.StockCheckModel.FindList(l.ctx, in.Page, in.PageSize, in.CheckNo, in.Status, in.StartDate, in.EndDate)
tenantId := tenantctx.ExtractTenantId(l.ctx)
list, total, err := l.svcCtx.StockCheckModel.FindList(l.ctx, tenantId, in.Page, in.PageSize, in.CheckNo, in.Status, in.StartDate, in.EndDate)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}

7
rpc/system/internal/logic/listconfiglogic.go
View File

@ -4,6 +4,7 @@ import (
"context"
"muyu-apiserver/model"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/system/internal/svc"
"muyu-apiserver/rpc/system/pb"
@ -27,13 +28,15 @@ func NewListConfigLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ListCo
}
func (l *ListConfigLogic) ListConfig(in *pb.ListConfigReq) (*pb.ListConfigResp, error) {
tenantId := tenantctx.ExtractTenantId(l.ctx)
var err error
var configs []*model.SysConfig
if in.ConfigGroup != "" {
configs, err = l.svcCtx.ConfigModel.FindByGroup(l.ctx, in.ConfigGroup)
configs, err = l.svcCtx.ConfigModel.FindByGroup(l.ctx, tenantId, in.ConfigGroup)
} else {
configs, err = l.svcCtx.ConfigModel.FindAll(l.ctx)
configs, err = l.svcCtx.ConfigModel.FindAll(l.ctx, tenantId)
}
if err != nil {
return nil, status.Error(codes.Internal, err.Error())

5
rpc/system/internal/logic/listrolelogic.go
View File

@ -3,6 +3,7 @@ package logic
import (
"context"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/system/internal/svc"
"muyu-apiserver/rpc/system/pb"
@ -26,7 +27,9 @@ func NewListRoleLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ListRole
}
func (l *ListRoleLogic) ListRole(in *pb.ListRoleReq) (*pb.ListRoleResp, error) {
roles, total, err := l.svcCtx.RoleModel.FindList(l.ctx, in.Page, in.PageSize, in.RoleName, in.Status)
tenantId := tenantctx.ExtractTenantId(l.ctx)
roles, total, err := l.svcCtx.RoleModel.FindList(l.ctx, tenantId, in.Page, in.PageSize, in.RoleName, in.Status)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}

5
rpc/system/internal/logic/listuserlogic.go
View File

@ -3,6 +3,7 @@ package logic
import (
"context"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/system/internal/svc"
"muyu-apiserver/rpc/system/pb"
@ -26,7 +27,9 @@ func NewListUserLogic(ctx context.Context, svcCtx *svc.ServiceContext) *ListUser
}
func (l *ListUserLogic) ListUser(in *pb.ListUserReq) (*pb.ListUserResp, error) {
users, total, err := l.svcCtx.UserModel.FindList(l.ctx, in.Page, in.PageSize, in.Username, in.RealName, in.Phone, in.Status)
tenantId := tenantctx.ExtractTenantId(l.ctx)
users, total, err := l.svcCtx.UserModel.FindList(l.ctx, tenantId, in.Page, in.PageSize, in.Username, in.RealName, in.Phone, in.Status)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}

5
rpc/system/internal/logic/updateconfiglogic.go
View File

@ -3,6 +3,7 @@ package logic
import (
"context"
"muyu-apiserver/pkg/tenantctx"
"muyu-apiserver/rpc/system/internal/svc"
"muyu-apiserver/rpc/system/pb"
@ -26,11 +27,13 @@ func NewUpdateConfigLogic(ctx context.Context, svcCtx *svc.ServiceContext) *Upda
}
func (l *UpdateConfigLogic) UpdateConfig(in *pb.UpdateConfigReq) (*pb.Empty, error) {
tenantId := tenantctx.ExtractTenantId(l.ctx)
if in.ConfigKey == "" {
return nil, status.Error(codes.InvalidArgument, "config_key is required")
}
err := l.svcCtx.ConfigModel.UpdateByKey(l.ctx, in.ConfigKey, in.ConfigValue)
err := l.svcCtx.ConfigModel.UpdateByKey(l.ctx, tenantId, in.ConfigKey, in.ConfigValue)
if err != nil {
return nil, status.Error(codes.Internal, err.Error())
}

29
worker/graphsync/go.mod Normal file
View File

@ -0,0 +1,29 @@
module muyu-apiserver/worker/graphsync
go 1.25.4
require (
github.com/IBM/sarama v1.43.3
github.com/neo4j/neo4j-go-driver/v5 v5.28.0
)
require (
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/eapache/go-resiliency v1.7.0 // indirect
github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
github.com/eapache/queue v1.1.0 // indirect
github.com/golang/snappy v0.0.4 // indirect
github.com/hashicorp/errwrap v1.0.0 // indirect
github.com/hashicorp/go-multierror v1.1.1 // indirect
github.com/hashicorp/go-uuid v1.0.3 // indirect
github.com/jcmturner/aescts/v2 v2.0.0 // indirect
github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
github.com/jcmturner/gofork v1.7.6 // indirect
github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
github.com/jcmturner/rpc/v2 v2.0.3 // indirect
github.com/klauspost/compress v1.17.9 // indirect
github.com/pierrec/lz4/v4 v4.1.21 // indirect
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
golang.org/x/crypto v0.26.0 // indirect
golang.org/x/net v0.28.0 // indirect
)

149
worker/graphsync/main.go Normal file
View File

@ -0,0 +1,149 @@
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"log"
"os"
"os/signal"
"strings"
"syscall"
"time"
"github.com/IBM/sarama"
"github.com/neo4j/neo4j-go-driver/v5/neo4j"
)
type Config struct {
KafkaBrokers string
KafkaTopic string
KafkaGroup string
Neo4jURI string
Neo4jUser string
Neo4jPass string
}
func main() {
var cfg Config
flag.StringVar(&cfg.KafkaBrokers, "kafka-brokers", envOr("KAFKA_BROKERS", "localhost:9092"), "Kafka brokers")
flag.StringVar(&cfg.KafkaTopic, "kafka-topic", envOr("KAFKA_TOPIC", "muyu_crm.public.crm_tenant_relationship"), "CDC topic")
flag.StringVar(&cfg.KafkaGroup, "kafka-group", envOr("KAFKA_GROUP", "graphsync"), "consumer group")
flag.StringVar(&cfg.Neo4jURI, "neo4j-uri", envOr("NEO4J_URI", "bolt://localhost:7687"), "Neo4j bolt URI")
flag.StringVar(&cfg.Neo4jUser, "neo4j-user", envOr("NEO4J_USER", "neo4j"), "Neo4j user")
flag.StringVar(&cfg.Neo4jPass, "neo4j-pass", envOr("NEO4J_PASS", "muyu2026"), "Neo4j password")
flag.Parse()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
driver, err := neo4j.NewDriverWithContext(cfg.Neo4jURI, neo4j.BasicAuth(cfg.Neo4jUser, cfg.Neo4jPass, ""))
if err != nil {
log.Fatalf("neo4j connect failed: %v", err)
}
defer driver.Close(ctx)
saramaCfg := sarama.NewConfig()
saramaCfg.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.NewBalanceStrategyRoundRobin()}
saramaCfg.Consumer.Offsets.Initial = sarama.OffsetOldest
group, err := sarama.NewConsumerGroup(strings.Split(cfg.KafkaBrokers, ","), cfg.KafkaGroup, saramaCfg)
if err != nil {
log.Fatalf("kafka consumer group failed: %v", err)
}
defer group.Close()
handler := &syncHandler{driver: driver}
go func() {
for {
if err := group.Consume(ctx, []string{cfg.KafkaTopic}, handler); err != nil {
log.Printf("consume error: %v, retrying in 5s...", err)
time.Sleep(5 * time.Second)
}
if ctx.Err() != nil {
return
}
}
}()
log.Printf("GraphSyncWorker started: brokers=%s topic=%s neo4j=%s", cfg.KafkaBrokers, cfg.KafkaTopic, cfg.Neo4jURI)
sig := make(chan os.Signal, 1)
signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
<-sig
log.Println("shutting down...")
cancel()
}
type syncHandler struct {
driver neo4j.DriverWithContext
}
func (h *syncHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil }
func (h *syncHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }
func (h *syncHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for msg := range claim.Messages() {
if err := h.processMessage(session.Context(), msg.Value); err != nil {
log.Printf("process msg offset=%d err: %v", msg.Offset, err)
}
session.MarkMessage(msg, "")
}
return nil
}
type cdcEvent struct {
Op string `json:"op"`
After map[string]interface{} `json:"after"`
}
func (h *syncHandler) processMessage(ctx context.Context, data []byte) error {
var evt cdcEvent
if err := json.Unmarshal(data, &evt); err != nil {
return fmt.Errorf("unmarshal: %w", err)
}
if evt.After == nil {
return nil
}
session := h.driver.NewSession(ctx, neo4j.SessionConfig{AccessMode: neo4j.AccessModeWrite})
defer session.Close(ctx)
fromTenant, _ := evt.After["from_tenant_id"].(string)
toTenant, _ := evt.After["to_tenant_id"].(string)
relType, _ := evt.After["relation_type"].(string)
relID, _ := evt.After["relationship_id"].(string)
if fromTenant == "" || toTenant == "" {
return nil
}
cypher := `
MERGE (a:Tenant {tenantId: $from})
MERGE (b:Tenant {tenantId: $to})
MERGE (a)-[r:SUPPLIES {relationshipId: $relId}]->(b)
SET r.relationType = $relType, r.updatedAt = datetime()
`
_, err := session.Run(ctx, cypher, map[string]interface{}{
"from": fromTenant,
"to": toTenant,
"relId": relID,
"relType": relType,
})
if err != nil {
return fmt.Errorf("neo4j exec: %w", err)
}
log.Printf("synced: %s -> %s [%s] id=%s", fromTenant, toTenant, relType, relID)
return nil
}
func envOr(key, fallback string) string {
if v := os.Getenv(key); v != "" {
return v
}
return fallback
}