From 1f4ccfb54e80c4361012f09c51c245890499732f Mon Sep 17 00:00:00 2001 From: Chever John Date: Mon, 30 Mar 2026 02:53:55 +0000 Subject: [PATCH] feat: multi-tenant CRM with PostgreSQL graph, tenant isolation, and Casbin RBAC - JWT claims extended with tenantId; login enforces strict tenant verification - AuthorityMiddleware: tenant scope check + Casbin path permission + anti-spoofing - CRM relation API (upstream/downstream one-hop, create/update/history, full graph) - CrmRepo backed by PostgreSQL with $N placeholders - gRPC tenant propagation via UnaryClientInterceptor (x-tenant-id metadata) - All legacy tables (12) gain tenant_id column with indexes - All model queries inject WHERE tenant_id filter - Casbin gorm-adapter downgraded to v3.28.0 for v2 compatibility - GraphSyncWorker (Kafka -> Neo4j) with idempotent MERGE - Full graph API restricted to admin role only - Database migrations for MySQL (CRM tables + tenant columns) and PostgreSQL (CRM init) - Docker Compose: added postgres service to main stack, graph stack with Kafka/Debezium/Neo4j Made-with: Cursor --- .gitignore | 2 + deploy/Dockerfile.graphsync | 8 + deploy/README-graph-stack.md | 35 +++ deploy/docker-compose.graph.yaml | 120 +++++++++ deploy/docker-compose.yaml | 21 ++ deploy/etc/gateway.yaml | 4 + deploy/mysql/init.sql | 101 ++++++- .../20260329_add_tenant_id_to_legacy.sql | 93 +++++++ .../migrations/20260329_multitenant_crm.sql | 87 ++++++ deploy/nginx/nginx.conf | 5 +- deploy/postgres/init-crm.sql | 70 +++++ deploy/postgres/init.sql | 105 ++++++++ gateway/etc/gateway-api.yaml | 7 + gateway/etc/gateway.yaml | 4 + gateway/gateway.api | 76 ++++++ gateway/gateway.go | 9 + gateway/internal/config/config.go | 11 +- .../crm/relation/createRelationHandler.go | 30 +++ .../crm/relation/getFullGraphHandler.go | 33 +++ .../crm/relation/listDownstreamHandler.go | 30 +++ .../relation/listRelationHistoryHandler.go | 30 +++ .../crm/relation/listUpstreamHandler.go | 30 +++ .../crm/relation/updateRelationHandler.go | 33 +++ gateway/internal/handler/routes.go | 41 +++ .../internal/logic/auth/getUserInfoLogic.go | 2 + gateway/internal/logic/auth/loginLogic.go | 13 + gateway/internal/logic/crm/relation/common.go | 32 +++ .../logic/crm/relation/createRelationLogic.go | 55 ++++ .../logic/crm/relation/listDownstreamLogic.go | 58 ++++ .../crm/relation/listRelationHistoryLogic.go | 56 ++++ .../logic/crm/relation/listUpstreamLogic.go | 58 ++++ .../logic/crm/relation/updateRelationLogic.go | 50 ++++ .../middleware/authorityMiddleware.go | 53 +++- gateway/internal/repo/crmrepo.go | 250 ++++++++++++++++++ gateway/internal/svc/serviceContext.go | 43 ++- gateway/internal/types/types.go | 56 ++++ go.mod | 5 +- go.sum | 10 +- model/invproductmodel.go | 32 +-- model/invstockadjustmodel.go | 8 +- model/invstockcheckmodel.go | 8 +- model/invstockimportlogmodel.go | 12 +- model/sysconfigmodel.go | 24 +- model/sysrolemodel.go | 16 +- model/sysusermodel.go | 8 +- pkg/ctxdata/ctxdata.go | 7 + pkg/jwt/jwt.go | 12 +- pkg/tenantctx/tenantctx.go | 35 +++ .../internal/logic/getstockgrouplogic.go | 6 +- .../internal/logic/getstocksummarylogic.go | 4 +- .../internal/logic/listimportloglogic.go | 4 +- .../internal/logic/listproductlogic.go | 4 +- .../internal/logic/liststockadjustlogic.go | 4 +- .../internal/logic/liststockchecklogic.go | 4 +- rpc/system/internal/logic/listconfiglogic.go | 7 +- rpc/system/internal/logic/listrolelogic.go | 5 +- rpc/system/internal/logic/listuserlogic.go | 5 +- .../internal/logic/updateconfiglogic.go | 5 +- worker/graphsync/go.mod | 29 ++ worker/graphsync/main.go | 149 +++++++++++ 60 files changed, 2020 insertions(+), 94 deletions(-) create mode 100644 deploy/Dockerfile.graphsync create mode 100644 deploy/README-graph-stack.md create mode 100644 deploy/docker-compose.graph.yaml create mode 100644 deploy/mysql/migrations/20260329_add_tenant_id_to_legacy.sql create mode 100644 deploy/mysql/migrations/20260329_multitenant_crm.sql create mode 100644 deploy/postgres/init-crm.sql create mode 100644 deploy/postgres/init.sql create mode 100644 gateway/internal/handler/crm/relation/createRelationHandler.go create mode 100644 gateway/internal/handler/crm/relation/getFullGraphHandler.go create mode 100644 gateway/internal/handler/crm/relation/listDownstreamHandler.go create mode 100644 gateway/internal/handler/crm/relation/listRelationHistoryHandler.go create mode 100644 gateway/internal/handler/crm/relation/listUpstreamHandler.go create mode 100644 gateway/internal/handler/crm/relation/updateRelationHandler.go create mode 100644 gateway/internal/logic/crm/relation/common.go create mode 100644 gateway/internal/logic/crm/relation/createRelationLogic.go create mode 100644 gateway/internal/logic/crm/relation/listDownstreamLogic.go create mode 100644 gateway/internal/logic/crm/relation/listRelationHistoryLogic.go create mode 100644 gateway/internal/logic/crm/relation/listUpstreamLogic.go create mode 100644 gateway/internal/logic/crm/relation/updateRelationLogic.go create mode 100644 gateway/internal/repo/crmrepo.go create mode 100644 pkg/tenantctx/tenantctx.go create mode 100644 worker/graphsync/go.mod create mode 100644 worker/graphsync/main.go diff --git a/.gitignore b/.gitignore index 98f0b58..5b97400 100644 --- a/.gitignore +++ b/.gitignore @@ -88,3 +88,5 @@ data/ .env .env.* *.pem +deploy/bin/ +worker/graphsync/go.sum diff --git a/deploy/Dockerfile.graphsync b/deploy/Dockerfile.graphsync new file mode 100644 index 0000000..866fa63 --- /dev/null +++ b/deploy/Dockerfile.graphsync @@ -0,0 +1,8 @@ +FROM alpine:3.19 + +RUN apk add --no-cache ca-certificates tzdata + +WORKDIR /app +COPY deploy/bin/graphsync . + +CMD ["./graphsync"] diff --git a/deploy/README-graph-stack.md b/deploy/README-graph-stack.md new file mode 100644 index 0000000..d0dcb93 --- /dev/null +++ b/deploy/README-graph-stack.md @@ -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. diff --git a/deploy/docker-compose.graph.yaml b/deploy/docker-compose.graph.yaml new file mode 100644 index 0000000..41fe1ca --- /dev/null +++ b/deploy/docker-compose.graph.yaml @@ -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" diff --git a/deploy/docker-compose.yaml b/deploy/docker-compose.yaml index 78c25d6..397e2a7 100644 --- a/deploy/docker-compose.yaml +++ b/deploy/docker-compose.yaml @@ -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 diff --git a/deploy/etc/gateway.yaml b/deploy/etc/gateway.yaml index 9613f02..4f4565f 100644 --- a/deploy/etc/gateway.yaml +++ b/deploy/etc/gateway.yaml @@ -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: diff --git a/deploy/mysql/init.sql b/deploy/mysql/init.sql index 19e972e..5ff5118 100644 --- a/deploy/mysql/init.sql +++ b/deploy/mysql/init.sql @@ -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'), diff --git a/deploy/mysql/migrations/20260329_add_tenant_id_to_legacy.sql b/deploy/mysql/migrations/20260329_add_tenant_id_to_legacy.sql new file mode 100644 index 0000000..861c1b2 --- /dev/null +++ b/deploy/mysql/migrations/20260329_add_tenant_id_to_legacy.sql @@ -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); diff --git a/deploy/mysql/migrations/20260329_multitenant_crm.sql b/deploy/mysql/migrations/20260329_multitenant_crm.sql new file mode 100644 index 0000000..4cc036d --- /dev/null +++ b/deploy/mysql/migrations/20260329_multitenant_crm.sql @@ -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' +); diff --git a/deploy/nginx/nginx.conf b/deploy/nginx/nginx.conf index 7035d8e..0382cb5 100644 --- a/deploy/nginx/nginx.conf +++ b/deploy/nginx/nginx.conf @@ -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; diff --git a/deploy/postgres/init-crm.sql b/deploy/postgres/init-crm.sql new file mode 100644 index 0000000..fd74b9e --- /dev/null +++ b/deploy/postgres/init-crm.sql @@ -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; diff --git a/deploy/postgres/init.sql b/deploy/postgres/init.sql new file mode 100644 index 0000000..c885d74 --- /dev/null +++ b/deploy/postgres/init.sql @@ -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; diff --git a/gateway/etc/gateway-api.yaml b/gateway/etc/gateway-api.yaml index 8144aae..2f154e5 100644 --- a/gateway/etc/gateway-api.yaml +++ b/gateway/etc/gateway-api.yaml @@ -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 diff --git a/gateway/etc/gateway.yaml b/gateway/etc/gateway.yaml index 54f9b17..ac9decd 100644 --- a/gateway/etc/gateway.yaml +++ b/gateway/etc/gateway.yaml @@ -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: diff --git a/gateway/gateway.api b/gateway/gateway.api index 700647a..1ad15af 100644 --- a/gateway/gateway.api +++ b/gateway/gateway.api @@ -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 diff --git a/gateway/gateway.go b/gateway/gateway.go index e002bdd..07b6af0 100644 --- a/gateway/gateway.go +++ b/gateway/gateway.go @@ -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 { diff --git a/gateway/internal/config/config.go b/gateway/internal/config/config.go index 9ae6a40..9e05208 100644 --- a/gateway/internal/config/config.go +++ b/gateway/internal/config/config.go @@ -11,9 +11,12 @@ import ( type Config struct { rest.RestConf Auth struct { - AccessSecret string - AccessExpire int64 + AccessSecret string + AccessExpire int64 + DefaultTenantId string } - SystemRpc zrpc.RpcClientConf - InventoryRpc zrpc.RpcClientConf + DataSource string + PostgresDataSource string + SystemRpc zrpc.RpcClientConf + InventoryRpc zrpc.RpcClientConf } diff --git a/gateway/internal/handler/crm/relation/createRelationHandler.go b/gateway/internal/handler/crm/relation/createRelationHandler.go new file mode 100644 index 0000000..65bd68c --- /dev/null +++ b/gateway/internal/handler/crm/relation/createRelationHandler.go @@ -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) + } + } +} + diff --git a/gateway/internal/handler/crm/relation/getFullGraphHandler.go b/gateway/internal/handler/crm/relation/getFullGraphHandler.go new file mode 100644 index 0000000..d178a8b --- /dev/null +++ b/gateway/internal/handler/crm/relation/getFullGraphHandler.go @@ -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) + } +} diff --git a/gateway/internal/handler/crm/relation/listDownstreamHandler.go b/gateway/internal/handler/crm/relation/listDownstreamHandler.go new file mode 100644 index 0000000..0f33454 --- /dev/null +++ b/gateway/internal/handler/crm/relation/listDownstreamHandler.go @@ -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) + } + } +} + diff --git a/gateway/internal/handler/crm/relation/listRelationHistoryHandler.go b/gateway/internal/handler/crm/relation/listRelationHistoryHandler.go new file mode 100644 index 0000000..082826e --- /dev/null +++ b/gateway/internal/handler/crm/relation/listRelationHistoryHandler.go @@ -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) + } + } +} + diff --git a/gateway/internal/handler/crm/relation/listUpstreamHandler.go b/gateway/internal/handler/crm/relation/listUpstreamHandler.go new file mode 100644 index 0000000..63e51ba --- /dev/null +++ b/gateway/internal/handler/crm/relation/listUpstreamHandler.go @@ -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) + } + } +} + diff --git a/gateway/internal/handler/crm/relation/updateRelationHandler.go b/gateway/internal/handler/crm/relation/updateRelationHandler.go new file mode 100644 index 0000000..9821ab8 --- /dev/null +++ b/gateway/internal/handler/crm/relation/updateRelationHandler.go @@ -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) + } + } +} + diff --git a/gateway/internal/handler/routes.go b/gateway/internal/handler/routes.go index 5e85ea3..2a2f0c8 100644 --- a/gateway/internal/handler/routes.go +++ b/gateway/internal/handler/routes.go @@ -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}, diff --git a/gateway/internal/logic/auth/getUserInfoLogic.go b/gateway/internal/logic/auth/getUserInfoLogic.go index c6b0484..8373c22 100644 --- a/gateway/internal/logic/auth/getUserInfoLogic.go +++ b/gateway/internal/logic/auth/getUserInfoLogic.go @@ -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, diff --git a/gateway/internal/logic/auth/loginLogic.go b/gateway/internal/logic/auth/loginLogic.go index e6c7886..93d9790 100644 --- a/gateway/internal/logic/auth/loginLogic.go +++ b/gateway/internal/logic/auth/loginLogic.go @@ -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 } diff --git a/gateway/internal/logic/crm/relation/common.go b/gateway/internal/logic/crm/relation/common.go new file mode 100644 index 0000000..ed83b42 --- /dev/null +++ b/gateway/internal/logic/crm/relation/common.go @@ -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} +} + diff --git a/gateway/internal/logic/crm/relation/createRelationLogic.go b/gateway/internal/logic/crm/relation/createRelationLogic.go new file mode 100644 index 0000000..a21d9c8 --- /dev/null +++ b/gateway/internal/logic/crm/relation/createRelationLogic.go @@ -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 +} + diff --git a/gateway/internal/logic/crm/relation/listDownstreamLogic.go b/gateway/internal/logic/crm/relation/listDownstreamLogic.go new file mode 100644 index 0000000..3cd1986 --- /dev/null +++ b/gateway/internal/logic/crm/relation/listDownstreamLogic.go @@ -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 +} + diff --git a/gateway/internal/logic/crm/relation/listRelationHistoryLogic.go b/gateway/internal/logic/crm/relation/listRelationHistoryLogic.go new file mode 100644 index 0000000..e4ec2dc --- /dev/null +++ b/gateway/internal/logic/crm/relation/listRelationHistoryLogic.go @@ -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 +} + diff --git a/gateway/internal/logic/crm/relation/listUpstreamLogic.go b/gateway/internal/logic/crm/relation/listUpstreamLogic.go new file mode 100644 index 0000000..135fab8 --- /dev/null +++ b/gateway/internal/logic/crm/relation/listUpstreamLogic.go @@ -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 +} + diff --git a/gateway/internal/logic/crm/relation/updateRelationLogic.go b/gateway/internal/logic/crm/relation/updateRelationLogic.go new file mode 100644 index 0000000..56e6f50 --- /dev/null +++ b/gateway/internal/logic/crm/relation/updateRelationLogic.go @@ -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 +} + diff --git a/gateway/internal/middleware/authorityMiddleware.go b/gateway/internal/middleware/authorityMiddleware.go index 15f5306..6bfe2cd 100644 --- a/gateway/internal/middleware/authorityMiddleware.go +++ b/gateway/internal/middleware/authorityMiddleware.go @@ -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) } } diff --git a/gateway/internal/repo/crmrepo.go b/gateway/internal/repo/crmrepo.go new file mode 100644 index 0000000..23ca64f --- /dev/null +++ b/gateway/internal/repo/crmrepo.go @@ -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 +} diff --git a/gateway/internal/svc/serviceContext.go b/gateway/internal/svc/serviceContext.go index 942b8b6..038c5c8 100644 --- a/gateway/internal/svc/serviceContext.go +++ b/gateway/internal/svc/serviceContext.go @@ -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()))), } } diff --git a/gateway/internal/types/types.go b/gateway/internal/types/types.go index 3eb33f5..778e55a 100644 --- a/gateway/internal/types/types.go +++ b/gateway/internal/types/types.go @@ -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"` +} diff --git a/go.mod b/go.mod index 6e06cf8..03eb75c 100644 --- a/go.mod +++ b/go.mod @@ -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 diff --git a/go.sum b/go.sum index 0e49ad0..5d2723b 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/model/invproductmodel.go b/model/invproductmodel.go index 0000cc1..961d758 100755 --- a/model/invproductmodel.go +++ b/model/invproductmodel.go @@ -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 } diff --git a/model/invstockadjustmodel.go b/model/invstockadjustmodel.go index 901993f..8e15dbe 100755 --- a/model/invstockadjustmodel.go +++ b/model/invstockadjustmodel.go @@ -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 ?" diff --git a/model/invstockcheckmodel.go b/model/invstockcheckmodel.go index c36f1e2..3ee3d56 100755 --- a/model/invstockcheckmodel.go +++ b/model/invstockcheckmodel.go @@ -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 ?" diff --git a/model/invstockimportlogmodel.go b/model/invstockimportlogmodel.go index 157be54..0be3cb9 100755 --- a/model/invstockimportlogmodel.go +++ b/model/invstockimportlogmodel.go @@ -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 } diff --git a/model/sysconfigmodel.go b/model/sysconfigmodel.go index f2f6dbb..7c56ac8 100755 --- a/model/sysconfigmodel.go +++ b/model/sysconfigmodel.go @@ -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 } diff --git a/model/sysrolemodel.go b/model/sysrolemodel.go index c2af91b..9ecbd0f 100755 --- a/model/sysrolemodel.go +++ b/model/sysrolemodel.go @@ -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 } diff --git a/model/sysusermodel.go b/model/sysusermodel.go index 22a74c1..62b1be9 100755 --- a/model/sysusermodel.go +++ b/model/sysusermodel.go @@ -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 ?" diff --git a/pkg/ctxdata/ctxdata.go b/pkg/ctxdata/ctxdata.go index eab89b6..a815e54 100644 --- a/pkg/ctxdata/ctxdata.go +++ b/pkg/ctxdata/ctxdata.go @@ -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 diff --git a/pkg/jwt/jwt.go b/pkg/jwt/jwt.go index 3b00e3f..47fea0c 100644 --- a/pkg/jwt/jwt.go +++ b/pkg/jwt/jwt.go @@ -7,16 +7,18 @@ import ( ) type Claims struct { - UserId string `json:"userId"` - RoleKey string `json:"roleKey"` + 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, + UserId: userId, + RoleKey: roleKey, + TenantId: tenantId, RegisteredClaims: jwt.RegisteredClaims{ ExpiresAt: jwt.NewNumericDate(now.Add(time.Duration(expire) * time.Second)), IssuedAt: jwt.NewNumericDate(now), diff --git a/pkg/tenantctx/tenantctx.go b/pkg/tenantctx/tenantctx.go new file mode 100644 index 0000000..8bfb269 --- /dev/null +++ b/pkg/tenantctx/tenantctx.go @@ -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...) + } +} diff --git a/rpc/inventory/internal/logic/getstockgrouplogic.go b/rpc/inventory/internal/logic/getstockgrouplogic.go index cc7cd63..908ee04 100644 --- a/rpc/inventory/internal/logic/getstockgrouplogic.go +++ b/rpc/inventory/internal/logic/getstockgrouplogic.go @@ -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) } diff --git a/rpc/inventory/internal/logic/getstocksummarylogic.go b/rpc/inventory/internal/logic/getstocksummarylogic.go index 0f2ee05..10825ed 100644 --- a/rpc/inventory/internal/logic/getstocksummarylogic.go +++ b/rpc/inventory/internal/logic/getstocksummarylogic.go @@ -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()) } diff --git a/rpc/inventory/internal/logic/listimportloglogic.go b/rpc/inventory/internal/logic/listimportloglogic.go index f07786b..aa0a6cf 100644 --- a/rpc/inventory/internal/logic/listimportloglogic.go +++ b/rpc/inventory/internal/logic/listimportloglogic.go @@ -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()) } diff --git a/rpc/inventory/internal/logic/listproductlogic.go b/rpc/inventory/internal/logic/listproductlogic.go index 06d1202..098e5f2 100644 --- a/rpc/inventory/internal/logic/listproductlogic.go +++ b/rpc/inventory/internal/logic/listproductlogic.go @@ -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()) } diff --git a/rpc/inventory/internal/logic/liststockadjustlogic.go b/rpc/inventory/internal/logic/liststockadjustlogic.go index 224252c..fb1dfff 100644 --- a/rpc/inventory/internal/logic/liststockadjustlogic.go +++ b/rpc/inventory/internal/logic/liststockadjustlogic.go @@ -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()) } diff --git a/rpc/inventory/internal/logic/liststockchecklogic.go b/rpc/inventory/internal/logic/liststockchecklogic.go index a8102de..8d0adda 100644 --- a/rpc/inventory/internal/logic/liststockchecklogic.go +++ b/rpc/inventory/internal/logic/liststockchecklogic.go @@ -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()) } diff --git a/rpc/system/internal/logic/listconfiglogic.go b/rpc/system/internal/logic/listconfiglogic.go index 3c37599..86f5a31 100644 --- a/rpc/system/internal/logic/listconfiglogic.go +++ b/rpc/system/internal/logic/listconfiglogic.go @@ -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()) diff --git a/rpc/system/internal/logic/listrolelogic.go b/rpc/system/internal/logic/listrolelogic.go index b1c8e7f..94c9399 100644 --- a/rpc/system/internal/logic/listrolelogic.go +++ b/rpc/system/internal/logic/listrolelogic.go @@ -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()) } diff --git a/rpc/system/internal/logic/listuserlogic.go b/rpc/system/internal/logic/listuserlogic.go index 1a2a7c2..0662300 100644 --- a/rpc/system/internal/logic/listuserlogic.go +++ b/rpc/system/internal/logic/listuserlogic.go @@ -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()) } diff --git a/rpc/system/internal/logic/updateconfiglogic.go b/rpc/system/internal/logic/updateconfiglogic.go index 5942abb..5e0cb3d 100644 --- a/rpc/system/internal/logic/updateconfiglogic.go +++ b/rpc/system/internal/logic/updateconfiglogic.go @@ -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()) } diff --git a/worker/graphsync/go.mod b/worker/graphsync/go.mod new file mode 100644 index 0000000..5ab448c --- /dev/null +++ b/worker/graphsync/go.mod @@ -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 +) diff --git a/worker/graphsync/main.go b/worker/graphsync/main.go new file mode 100644 index 0000000..08f4fed --- /dev/null +++ b/worker/graphsync/main.go @@ -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 +}