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 }