package handler

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	"github.com/wmms/wmms/internal/model"
	"github.com/wmms/wmms/internal/testutil"
	"golang.org/x/crypto/bcrypt"
)

// These tests add real route/auth/organization-boundary coverage. Existing
// repository/service suites own inventory concurrency and insufficient-stock
// rollback; this suite adds concurrent approval and mid-transaction failures.
// Do not parallelize: testutil swaps the process-global database.DB.
type companyHTTPFixture struct {
	db     *sql.DB
	router *gin.Engine
	tokens map[string]string
}

func companySQL(t *testing.T, db *sql.DB, query string, args ...interface{}) {
	t.Helper()
	if _, err := db.Exec(query, args...); err != nil {
		t.Fatalf("fixture SQL: %v", err)
	}
}

func newCompanyHTTPFixture(t *testing.T) *companyHTTPFixture {
	t.Helper()
	// Requires TEST_DB_DSN with company_isolation_test prefix; creates and drops
	// only a generated disposable schema. An unset DSN is a skip, not acceptance.
	db := testutil.NewIsolationDB(t)
	companySQL(t, db, `INSERT INTO organizations(id,code,name) VALUES(301,'HTTP_BJ','北京公司'),(302,'HTTP_WH','武汉公司');
 INSERT INTO users(id,username,password) VALUES(301,'http-beijing','x'),(302,'http-wuhan','x'),(303,'http-both','x'),(304,'http-admin','x');
 INSERT INTO user_roles(user_id,role_id) SELECT u.id,r.id FROM users u CROSS JOIN roles r WHERE u.id IN(301,302,303) AND r.code='warehouse_admin';
 INSERT INTO user_roles(user_id,role_id) SELECT 304,id FROM roles WHERE code='super_admin';
 INSERT INTO user_organizations(user_id,organization_id,is_default) VALUES(301,301,1),(302,302,1),(303,301,1),(303,302,0);
 INSERT INTO material_categories(id,organization_id,code,name) VALUES(301,301,'BJ','北京机械'),(302,302,'WH','武汉电气');
 INSERT INTO material_categories(id,organization_id,code,name,parent_id) VALUES(311,301,'BJ01','北京标准件',301),(312,302,'WH01','武汉元器件',302);
 INSERT INTO units(id,organization_id,code,name) VALUES(301,301,'PCS','件'),(302,302,'KG','千克');
 INSERT INTO code_rules(id,organization_id,category_id,prefix,sequence_length,next_sequence) VALUES(301,301,311,'BJ',4,1),(302,302,312,'WH',4,1);
 INSERT INTO materials(id,organization_id,material_code,name,spec,unit,category_id,base_unit_id) VALUES(301,301,'SHARED','北京同编码物料','','件',311,301),(302,302,'SHARED','武汉同编码物料','','千克',312,302);
 INSERT INTO warehouses(id,organization_id,code,name) VALUES(301,301,'BJ_W','北京仓库'),(302,302,'WH_W','武汉仓库');
 INSERT INTO warehouse_locations(id,warehouse_id,code,name) VALUES(301,301,'BJ_L','北京库位'),(302,302,'WH_L','武汉库位');`)
	password, err := bcrypt.GenerateFromPassword([]byte("CompanyTest123!"), bcrypt.MinCost)
	if err != nil {
		t.Fatal(err)
	}
	companySQL(t, db, `UPDATE users SET password=? WHERE id BETWEEN 301 AND 304`, string(password))
	gin.SetMode(gin.TestMode)
	f := &companyHTTPFixture{db: db, router: gin.New(), tokens: map[string]string{}}
	NewHandler().RegisterRoutes(f.router)
	for _, name := range []string{"beijing", "wuhan", "both", "admin"} {
		var login struct {
			Token string `json:"token"`
		}
		data := f.request(t, "", "", "POST", "/api/auth/login", fmt.Sprintf(`{"username":"http-%s","password":"CompanyTest123!"}`, name), 200)
		companyDecode(t, data, &login)
		if login.Token == "" {
			t.Fatalf("login %s returned no token", name)
		}
		f.tokens[name] = login.Token
	}
	return f
}

func (f *companyHTTPFixture) serve(account, org, method, path, body string) *httptest.ResponseRecorder {
	req := httptest.NewRequest(method, path, strings.NewReader(body))
	if account != "" {
		req.Header.Set("Authorization", "Bearer "+f.tokens[account])
	}
	if org != "" {
		req.Header.Set("X-Organization-ID", org)
	}
	if body != "" {
		req.Header.Set("Content-Type", "application/json")
	}
	w := httptest.NewRecorder()
	f.router.ServeHTTP(w, req)
	return w
}

func (f *companyHTTPFixture) request(t *testing.T, account, org, method, path, body string, want int) json.RawMessage {
	t.Helper()
	w := f.serve(account, org, method, path, body)
	if w.Code != want {
		t.Fatalf("%s %s account=%s org=%s: status=%d want=%d body=%s", method, path, account, org, w.Code, want, w.Body.String())
	}
	var envelope struct {
		Code int             `json:"code"`
		Data json.RawMessage `json:"data"`
	}
	companyDecode(t, w.Body.Bytes(), &envelope)
	if want < 300 && envelope.Code != 0 {
		t.Fatalf("business error: %s", w.Body.String())
	}
	return envelope.Data
}

func companyDecode(t *testing.T, data []byte, value interface{}) {
	t.Helper()
	if err := json.Unmarshal(data, value); err != nil {
		t.Fatalf("decode %s: %v", data, err)
	}
}

type companyPage[T any] struct {
	Items   []T   `json:"items"`
	Total   int64 `json:"total"`
	HasMore bool  `json:"has_more"`
}

func companyGetPage[T any](t *testing.T, f *companyHTTPFixture, account string, org int64, path string, total int64) companyPage[T] {
	t.Helper()
	var page companyPage[T]
	companyDecode(t, f.request(t, account, fmt.Sprint(org), "GET", path, "", 200), &page)
	if page.Total != total {
		t.Fatalf("%s org=%d total=%d want=%d items=%+v", path, org, page.Total, total, page.Items)
	}
	return page
}

func TestCompanyIsolationAuthorizationMySQL(t *testing.T) {
	f := newCompanyHTTPFixture(t)
	for _, tc := range []struct {
		account string
		org     int64
		want    int
	}{
		{"beijing", 301, 200}, {"beijing", 302, 403}, {"wuhan", 301, 403}, {"wuhan", 302, 200},
		{"both", 301, 200}, {"both", 302, 200}, {"admin", 301, 200}, {"admin", 302, 200}, {"admin", 999999, 403},
	} {
		t.Run(fmt.Sprintf("%s/%d", tc.account, tc.org), func(t *testing.T) { f.request(t, tc.account, fmt.Sprint(tc.org), "GET", "/api/materials", "", tc.want) })
	}
	for _, header := range []string{"", "0", "-1", "abc", "301.5"} {
		t.Run("invalid-header/"+header, func(t *testing.T) { f.request(t, "admin", header, "GET", "/api/units", "", 400) })
	}
	f.request(t, "", "301", "GET", "/api/materials", "", 401)
	// Organization grants do not grant admin configuration or approval roles.
	f.request(t, "both", "301", "POST", "/api/units", `{"code":"NEW","name":"新单位"}`, 403)
	f.request(t, "both", "301", "GET", "/api/applications", "", 401)
	for _, tc := range []struct {
		account string
		ids     []int64
		def     int64
	}{
		{"beijing", []int64{301}, 301}, {"wuhan", []int64{302}, 302}, {"both", []int64{301, 302}, 301}, {"admin", []int64{1, 301, 302}, 0},
	} {
		var access model.OrganizationAccess
		companyDecode(t, f.request(t, tc.account, "", "GET", "/api/auth/organizations", "", 200), &access)
		if len(access.Organizations) != len(tc.ids) {
			t.Fatalf("%s access=%+v", tc.account, access)
		}
		for _, want := range tc.ids {
			found := false
			for _, org := range access.Organizations {
				found = found || org.ID == want
			}
			if !found {
				t.Fatalf("%s missing company %d", tc.account, want)
			}
		}
		if tc.def != 0 && (access.DefaultOrganizationID == nil || *access.DefaultOrganizationID != tc.def) {
			t.Fatalf("%s default=%v", tc.account, access.DefaultOrganizationID)
		}
	}
	for _, target := range []int64{302, 301} {
		f.request(t, "both", "", "POST", "/api/auth/organization-context", fmt.Sprintf(`{"organization_id":%d,"previous_organization_id":%d}`, target, 603-target), 200)
	}
	var audits, defaults int
	if err := f.db.QueryRow(`SELECT COUNT(*) FROM organization_audit_logs WHERE actor_user_id=303`).Scan(&audits); err != nil {
		t.Fatal(err)
	}
	if err := f.db.QueryRow(`SELECT COUNT(*) FROM user_organizations WHERE user_id=303 AND organization_id=301 AND is_default=1`).Scan(&defaults); err != nil || defaults != 1 || audits != 2 {
		t.Fatalf("switch audits=%d default=%d err=%v", audits, defaults, err)
	}
	// Database revocation and stale JWT roles remain in the existing context suite.
}

func TestCompanyIsolationMasterDataMySQL(t *testing.T) {
	f := newCompanyHTTPFixture(t)
	for _, org := range []int64{301, 302} {
		for _, tc := range []struct {
			path string
			want int
		}{{"/api/material-categories", 2}, {"/api/units", 1}, {"/api/code-rules", 1}} {
			var items []struct {
				OrganizationID int64 `json:"organization_id"`
			}
			companyDecode(t, f.request(t, "both", fmt.Sprint(org), "GET", tc.path, "", 200), &items)
			if len(items) != tc.want {
				t.Fatalf("%s count=%d want=%d", tc.path, len(items), tc.want)
			}
			for _, item := range items {
				if item.OrganizationID != org {
					t.Fatalf("%s leaked %+v", tc.path, item)
				}
			}
		}
		page := companyGetPage[model.Material](t, f, "both", org, "/api/materials?search=SHARED&limit=1", 1)
		if len(page.Items) != 1 || page.Items[0].ID != org || page.Items[0].OrganizationID != org || page.HasMore {
			t.Fatalf("same-code search org=%d: %+v", org, page)
		}
		var detail model.Material
		companyDecode(t, f.request(t, "both", fmt.Sprint(org), "GET", fmt.Sprintf("/api/materials/%d", org), "", 200), &detail)
		if detail.CategoryID == nil || *detail.CategoryID != org+10 || detail.BaseUnitID == nil || *detail.BaseUnitID != org {
			t.Fatalf("references=%+v", detail)
		}
		f.request(t, "both", fmt.Sprint(org), "GET", fmt.Sprintf("/api/materials/%d", 603-org), "", 404)
	}
	preview := func(org int64, code string) {
		t.Helper()
		var next model.NextMaterialCode
		companyDecode(t, f.request(t, "both", fmt.Sprint(org), "GET", fmt.Sprintf("/api/code-rules/next?category_id=%d", org+10), "", 200), &next)
		if !next.Available || next.Code != code {
			t.Fatalf("org=%d preview=%+v want=%s", org, next, code)
		}
	}
	create := func(org int64, code string) model.Material {
		t.Helper()
		var m model.Material
		body := fmt.Sprintf(`{"organization_id":%d,"material_code":%q,"name":"路由建档","category_id":%d,"base_unit_id":%d}`, 603-org, code, org+10, org)
		companyDecode(t, f.request(t, "both", fmt.Sprint(org), "POST", "/api/materials", body, 201), &m)
		if m.OrganizationID != org {
			t.Fatalf("forged ownership=%+v", m)
		}
		return m
	}
	preview(301, "BJ0001")
	preview(302, "WH0001")
	a := create(301, "BJ0001")
	preview(301, "BJ0002")
	preview(302, "WH0001")
	create(302, "WH0001")
	preview(302, "WH0002")
	create(301, "SHARED-NEW")
	create(302, "SHARED-NEW")
	f.request(t, "both", "301", "POST", "/api/materials", `{"material_code":"SHARED-NEW","name":"重复","category_id":311,"base_unit_id":301}`, 400)
	page := companyGetPage[model.Material](t, f, "both", 301, "/api/materials?limit=1", 3)
	if len(page.Items) != 1 || !page.HasMore {
		t.Fatalf("page=%+v", page)
	}
	page = companyGetPage[model.Material](t, f, "both", 301, "/api/materials?limit=1&offset=3", 3)
	if len(page.Items) != 0 || page.HasMore {
		t.Fatalf("last page=%+v", page)
	}
	companyGetPage[model.Material](t, f, "both", 301, "/api/materials?search=武汉", 0)
	for _, tc := range []struct{ path, body string }{
		{"/api/materials", `{"material_code":"BJ0002","name":"坏分类","category_id":312,"base_unit_id":301}`},
		{"/api/materials", `{"material_code":"BJ0002","name":"坏单位","category_id":311,"base_unit_id":302}`},
		{"/api/material-categories", `{"code":"BAD","name":"坏父级","parent_id":302}`},
		{"/api/code-rules", `{"category_id":312,"prefix":"BAD","sequence_length":4}`},
	} {
		f.request(t, "admin", "301", "POST", tc.path, tc.body, 400)
	}
	preview(301, "BJ0002")
	preview(302, "WH0002")
	companyGetPage[model.Material](t, f, "both", 301, "/api/materials", 3)
	for _, body := range []string{`{"category_id":312,"name":"坏分类"}`, `{"category_id":311,"base_unit_id":302,"name":"坏单位"}`} {
		f.request(t, "both", "301", "PUT", fmt.Sprintf("/api/materials/%d", a.ID), body, 400)
	}
	for _, tc := range []struct{ method, suffix, body string }{{"PUT", "", `{"category_id":311,"name":"恶意修改"}`}, {"POST", "/obsolete", ""}, {"POST", "/restore", ""}} {
		f.request(t, "both", "301", tc.method, "/api/materials/302"+tc.suffix, tc.body, 404)
	}
	f.request(t, "both", "301", "PUT", fmt.Sprintf("/api/materials/%d", a.ID), `{"organization_id":302,"category_id":311,"base_unit_id":301,"name":"北京更新"}`, 200)
	f.request(t, "both", "301", "POST", fmt.Sprintf("/api/materials/%d/obsolete", a.ID), "", 200)
	companyGetPage[model.Material](t, f, "both", 301, "/api/materials?status=obsolete", 1)
	companyGetPage[model.Material](t, f, "both", 302, "/api/materials?status=obsolete", 0)
	f.request(t, "both", "301", "POST", fmt.Sprintf("/api/materials/%d/restore", a.ID), "", 200)
	var updated model.Material
	companyDecode(t, f.request(t, "both", "301", "GET", fmt.Sprintf("/api/materials/%d", a.ID), "", 200), &updated)
	if updated.OrganizationID != 301 || updated.Name != "北京更新" || updated.Status != "active" || updated.Version != 4 {
		t.Fatalf("CRUD=%+v", updated)
	}
	var logs []model.MaterialLog
	companyDecode(t, f.request(t, "both", "301", "GET", fmt.Sprintf("/api/logs/materials/%d", a.ID), "", 200), &logs)
	if len(logs) != 4 {
		t.Fatalf("CRUD logs=%+v", logs)
	}
	for _, log := range logs {
		if log.MaterialID != a.ID {
			t.Fatalf("foreign log=%+v", log)
		}
	}
	companyGetPage[model.MaterialLog](t, f, "both", 301, "/api/logs?material_id=302", 0)
	f.request(t, "both", "301", "GET", "/api/logs/materials/302", "", 404)
	companyGetPage[model.MaterialLog](t, f, "both", 301, "/api/logs?limit=1", 5)
	companyGetPage[model.MaterialLog](t, f, "both", 302, "/api/logs", 2)
	// Configuration CRUD also takes ownership from the selected company.
	for _, org := range []int64{301, 302} {
		header := fmt.Sprint(org)
		var category model.MaterialCategory
		companyDecode(t, f.request(t, "admin", header, "POST", "/api/material-categories",
			fmt.Sprintf(`{"organization_id":%d,"code":"EXTRA","name":"新增小类","parent_id":%d}`, 603-org, org), 201), &category)
		f.request(t, "admin", header, "PUT", fmt.Sprintf("/api/material-categories/%d", category.ID),
			fmt.Sprintf(`{"organization_id":%d,"name":"更新小类","parent_id":%d}`, 603-org, org), 200)
		var unit model.Unit
		companyDecode(t, f.request(t, "admin", header, "POST", "/api/units",
			fmt.Sprintf(`{"organization_id":%d,"code":"EXTRA","name":"新增单位"}`, 603-org), 201), &unit)
		// A no-op update is valid; MySQL must count the matched row as production does.
		f.request(t, "admin", header, "PUT", fmt.Sprintf("/api/units/%d", unit.ID),
			`{"name":"新增单位","precision_scale":0,"status":"active"}`, 200)
		f.request(t, "admin", header, "PUT", fmt.Sprintf("/api/units/%d", unit.ID),
			fmt.Sprintf(`{"organization_id":%d,"name":"更新单位","precision_scale":2}`, 603-org), 200)
		var rule model.CodeRule
		companyDecode(t, f.request(t, "admin", header, "POST", "/api/code-rules",
			fmt.Sprintf(`{"organization_id":%d,"category_id":%d,"prefix":"EX","sequence_length":3}`, 603-org, category.ID), 201), &rule)
		f.request(t, "admin", header, "PUT", fmt.Sprintf("/api/code-rules/%d", rule.ID),
			fmt.Sprintf(`{"organization_id":%d,"category_id":%d,"prefix":"EX","sequence_length":3,"next_sequence":9}`, 603-org, category.ID), 200)
		for _, tc := range []struct {
			table string
			id    int64
		}{
			{"material_categories", category.ID}, {"units", unit.ID}, {"code_rules", rule.ID},
		} {
			var owner int64
			if err := f.db.QueryRow("SELECT organization_id FROM "+tc.table+" WHERE id=?", tc.id).Scan(&owner); err != nil || owner != org {
				t.Fatalf("%s id=%d owner=%d want=%d err=%v", tc.table, tc.id, owner, org, err)
			}
		}
		var savedRule model.NextMaterialCode
		companyDecode(t, f.request(t, "both", header, "GET", fmt.Sprintf("/api/code-rules/next?category_id=%d", category.ID), "", 200), &savedRule)
		if !savedRule.Available || savedRule.Code != "EX009" {
			t.Fatalf("saved rule=%+v", savedRule)
		}
	}
	for _, tc := range []struct{ path, body string }{
		{"/api/material-categories/312", `{"name":"跨公司","parent_id":301}`},
		{"/api/units/302", `{"name":"跨公司"}`},
		{"/api/code-rules/302", `{"category_id":311,"prefix":"HACK","sequence_length":4}`},
	} {
		f.request(t, "admin", "301", "PUT", tc.path, tc.body, 404)
	}
}

func TestCompanyIsolationWarehouseApplicationFlowMySQL(t *testing.T) {
	f := newCompanyHTTPFixture(t)
	type flowIDs struct{ inbound, outbound, application int64 }
	flows := map[int64]flowIDs{}
	for _, org := range []int64{301, 302} {
		account, received, requested := "beijing", 10, 3
		if org == 302 {
			account, received, requested = "wuhan", 20, 4
		}
		header := fmt.Sprint(org)
		var in model.InboundOrder
		body := fmt.Sprintf(`{"organization_id":%d,"warehouse_id":%d,"location_id":%d,"handler":"经办","supplier":"供应商","items":[{"material_id":%d,"quantity":%d}]}`, 603-org, 603-org, 603-org, org, received)
		companyDecode(t, f.request(t, account, header, "POST", "/api/inbound", body, 201), &in)
		if in.OrganizationID != org || in.WarehouseID != org || in.LocationID != org {
			t.Fatalf("inbound scope=%+v", in)
		}
		var app model.MaterialApplication
		body = fmt.Sprintf(`{"organization_id":%d,"purpose":"领用","items":[{"material_id":%d,"quantity":%d}]}`, 603-org, org, requested)
		companyDecode(t, f.request(t, account, header, "POST", "/api/applications", body, 201), &app)
		if app.OrganizationID != org {
			t.Fatalf("application scope=%+v", app)
		}
		f.request(t, account, header, "POST", fmt.Sprintf("/api/applications/%d/approve", app.ID), `{"auto_outbound":true}`, 401)
		f.request(t, "admin", header, "POST", fmt.Sprintf("/api/applications/%d/approve", app.ID), `{"auto_outbound":true}`, 200)
		page := companyGetPage[model.OutboundOrder](t, f, account, org, "/api/outbound?limit=1", 1)
		if len(page.Items) != 1 || page.Items[0].OrganizationID != org || page.Items[0].WarehouseID != org || page.Items[0].LocationID != org {
			t.Fatalf("auto outbound=%+v", page)
		}
		var detail model.OutboundOrder
		companyDecode(t, f.request(t, account, header, "GET", fmt.Sprintf("/api/outbound/%d", page.Items[0].ID), "", 200), &detail)
		if len(detail.Items) != 1 || detail.Items[0].MaterialID != org || detail.Items[0].Quantity != requested {
			t.Fatalf("auto outbound items=%+v", detail)
		}
		flows[org] = flowIDs{in.ID, page.Items[0].ID, app.ID}
		stocks := companyGetPage[model.Stock](t, f, account, org, "/api/stock?search=SHARED&limit=1", 1)
		if len(stocks.Items) != 1 || stocks.Items[0].MaterialID != org || stocks.Items[0].TotalOnHand.String() != fmt.Sprint(received-requested) {
			t.Fatalf("stock=%+v want=%d", stocks, received-requested)
		}
		companyGetPage[model.InboundOrder](t, f, account, org, "/api/inbound", 1)
		companyGetPage[model.MaterialApplication](t, f, account, org, "/api/applications/my?status=approved", 1)
		companyGetPage[model.MaterialApplication](t, f, "both", org, "/api/applications/my", 0)
		companyGetPage[model.MaterialApplication](t, f, "admin", org, "/api/applications?status=approved", 1)
		f.request(t, account, header, "PUT", fmt.Sprintf("/api/stock/%d/warning", org), `{"warning_stock":100}`, 200)
		companyGetPage[model.Stock](t, f, account, org, "/api/stock/warning", 1)
		logs := companyGetPage[model.StockLogResponse](t, f, account, org, "/api/stock/logs?limit=1", 2)
		if len(logs.Items) != 1 || logs.Items[0].MaterialID != org || !logs.HasMore {
			t.Fatalf("stock logs=%+v", logs)
		}
		companyGetPage[model.StockLogResponse](t, f, account, org, fmt.Sprintf("/api/stock/logs/%d", org), 2)
		var dimensions []model.InventoryBalanceDetail
		companyDecode(t, f.request(t, account, header, "GET", fmt.Sprintf("/api/stock/%d/details", org), "", 200), &dimensions)
		if len(dimensions) != 1 || dimensions[0].OrganizationID != org || dimensions[0].WarehouseID != org || dimensions[0].LocationID != org {
			t.Fatalf("inventory dimensions=%+v", dimensions)
		}
		category := "BJ"
		if org == 302 {
			category = "WH"
		}
		companyGetPage[model.Stock](t, f, account, org, "/api/stock?category="+category, 1)
		companyGetPage[model.Stock](t, f, account, org, "/api/stock?category=OTHER", 0)
		var mismatches int
		if err := f.db.QueryRow(`SELECT COUNT(*) FROM stock_logs WHERE organization_id=? AND (warehouse_id<>? OR location_id<>? OR material_id<>?)`, org, org, org, org).Scan(&mismatches); err != nil || mismatches != 0 {
			t.Fatalf("stock dimensions mismatch=%d err=%v", mismatches, err)
		}
	}
	for _, org := range []int64{301, 302} {
		other := 603 - org
		ids := flows[other]
		for _, path := range []string{
			fmt.Sprintf("/api/materials/%d", other), fmt.Sprintf("/api/stock/%d", other), fmt.Sprintf("/api/stock/%d/details", other),
			fmt.Sprintf("/api/stock/logs/%d", other), fmt.Sprintf("/api/inbound/%d", ids.inbound), fmt.Sprintf("/api/outbound/%d", ids.outbound),
			fmt.Sprintf("/api/inbound/materials/%d", other), fmt.Sprintf("/api/outbound/materials/%d", other),
		} {
			f.request(t, "both", fmt.Sprint(org), "GET", path, "", 404)
		}
		f.request(t, "both", fmt.Sprint(org), "PUT", fmt.Sprintf("/api/stock/%d/warning", other), `{"warning_stock":99}`, 404)
		for _, suffix := range []string{"approve", "reject", "cancel"} {
			f.request(t, "admin", fmt.Sprint(org), "POST", fmt.Sprintf("/api/applications/%d/%s", ids.application, suffix), `{"auto_outbound":true,"reason":"跨公司"}`, 404)
		}
		for _, kind := range []string{"inbound", "outbound"} {
			id := ids.inbound
			if kind == "outbound" {
				id = ids.outbound
			}
			for _, method := range []string{"PUT", "DELETE"} {
				body := ""
				if method == "PUT" {
					body = fmt.Sprintf(`{"handler":"h","supplier":"s","receiver":"r","items":[{"material_id":%d,"quantity":1}]}`, org)
				}
				f.request(t, "both", fmt.Sprint(org), method, fmt.Sprintf("/api/%s/%d", kind, id), body, 404)
			}
		}
		f.request(t, "both", fmt.Sprint(org), "POST", "/api/applications", fmt.Sprintf(`{"items":[{"material_id":%d,"quantity":1}]}`, other), 400)
		for _, action := range []string{"cancel", "reject"} {
			var app model.MaterialApplication
			companyDecode(t, f.request(t, "both", fmt.Sprint(org), "POST", "/api/applications", fmt.Sprintf(`{"items":[{"material_id":%d,"quantity":1}]}`, org), 201), &app)
			account := "both"
			if action == "reject" {
				account = "admin"
			}
			f.request(t, account, fmt.Sprint(org), "POST", fmt.Sprintf("/api/applications/%d/%s", app.ID, action), `{"reason":"不需要"}`, 200)
			status := "cancelled"
			if action == "reject" {
				status = "rejected"
			}
			companyGetPage[model.MaterialApplication](t, f, "both", org, "/api/applications/my?status="+status, 1)
		}
		var balance string
		want := "7.000000"
		if org == 302 {
			want = "16.000000"
		}
		if err := f.db.QueryRow(`SELECT on_hand FROM inventory_balances WHERE organization_id=? AND warehouse_id=? AND location_id=? AND material_id=?`, org, org, org, org).Scan(&balance); err != nil || balance != want {
			t.Fatalf("final org=%d balance=%s want=%s err=%v", org, balance, want, err)
		}
		companyGetPage[model.OutboundOrder](t, f, "both", org, "/api/outbound", 1)
		// Existing order_flow integration owns inbound update/delete. Exercise the
		// corresponding manual outbound reversal through the real HTTP route.
		var manual model.OutboundOrder
		companyDecode(t, f.request(t, "both", fmt.Sprint(org), "POST", "/api/outbound",
			fmt.Sprintf(`{"receiver":"r","items":[{"material_id":%d,"quantity":1}]}`, org), 201), &manual)
		f.request(t, "both", fmt.Sprint(org), "PUT", fmt.Sprintf("/api/outbound/%d", manual.ID),
			fmt.Sprintf(`{"receiver":"r","items":[{"material_id":%d,"quantity":2}]}`, org), 200)
		f.request(t, "both", fmt.Sprint(org), "DELETE", fmt.Sprintf("/api/outbound/%d", manual.ID), "", 200)
		if err := f.db.QueryRow(`SELECT on_hand FROM inventory_balances WHERE organization_id=? AND material_id=?`, org, org).Scan(&balance); err != nil || balance != want {
			t.Fatalf("outbound reversal org=%d balance=%s want=%s err=%v", org, balance, want, err)
		}
		companyGetPage[model.OutboundOrder](t, f, "both", org, "/api/outbound", 1)
		for _, state := range []struct{ setup, restore string }{
			{"UPDATE warehouses SET status='inactive' WHERE id=?", "UPDATE warehouses SET status='active' WHERE id=?"},
			{"UPDATE warehouse_locations SET status='inactive' WHERE id=?", "UPDATE warehouse_locations SET status='active' WHERE id=?"},
		} {
			companySQL(t, f.db, state.setup, org)
			// Disabled dimensions still permit reading existing stock/documents.
			companyGetPage[model.Stock](t, f, "both", org, "/api/stock", 1)
			companyGetPage[model.InboundOrder](t, f, "both", org, "/api/inbound", 1)
			companyGetPage[model.OutboundOrder](t, f, "both", org, "/api/outbound", 1)
			f.request(t, "both", fmt.Sprint(org), "GET", fmt.Sprintf("/api/stock/%d/details", org), "", 200)
			own := flows[org]
			for _, req := range []struct{ method, path, body string }{
				{"PUT", fmt.Sprintf("/api/inbound/%d", own.inbound), fmt.Sprintf(`{"handler":"h","supplier":"s","items":[{"material_id":%d,"quantity":10}]}`, org)},
				{"DELETE", fmt.Sprintf("/api/inbound/%d", own.inbound), ""},
				{"PUT", fmt.Sprintf("/api/outbound/%d", own.outbound), fmt.Sprintf(`{"receiver":"r","items":[{"material_id":%d,"quantity":1}]}`, org)},
				{"DELETE", fmt.Sprintf("/api/outbound/%d", own.outbound), ""},
			} {
				f.request(t, "both", fmt.Sprint(org), req.method, req.path, req.body, 409)
			}
			companySQL(t, f.db, state.restore, org)
		}
	}
}

func TestCompanyIsolationInventoryConfigurationMySQL(t *testing.T) {
	f := newCompanyHTTPFixture(t)
	companySQL(t, f.db, `DELETE FROM warehouse_locations WHERE id=301; DELETE FROM warehouses WHERE id=301`)
	for _, tc := range []struct{ name, setup, message string }{
		{"no warehouse", "", "请先配置本公司仓库"},
		{"no location", `INSERT INTO warehouses(id,organization_id,code,name) VALUES(301,301,'BJ_W','北京仓库')`, "请先配置本公司仓库的库位"},
		{"inactive warehouse", `INSERT INTO warehouse_locations(id,warehouse_id,code,name) VALUES(301,301,'BJ_L','北京库位'); UPDATE warehouses SET status='inactive' WHERE id=301`, "本公司仓库已停用"},
		{"inactive location", `UPDATE warehouses SET status='active' WHERE id=301; UPDATE warehouse_locations SET status='inactive' WHERE id=301`, "本公司库位已停用"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			if tc.setup != "" {
				companySQL(t, f.db, tc.setup)
			}
			companyGetPage[model.Stock](t, f, "beijing", 301, "/api/stock", 0)
			f.request(t, "beijing", "301", "GET", "/api/materials/301", "", 200)
			var app model.MaterialApplication
			companyDecode(t, f.request(t, "beijing", "301", "POST", "/api/applications", `{"items":[{"material_id":301,"quantity":1}]}`, 201), &app)
			for _, req := range []struct{ method, path, body string }{
				{"POST", "/api/inbound", `{"handler":"h","supplier":"s","items":[{"material_id":301,"quantity":1}]}`},
				{"POST", "/api/outbound", `{"receiver":"r","items":[{"material_id":301,"quantity":1}]}`},
				{"PUT", "/api/stock/301/warning", `{"warning_stock":1}`},
				{"POST", fmt.Sprintf("/api/applications/%d/approve", app.ID), `{"auto_outbound":true}`},
			} {
				w := f.serve("admin", "301", req.method, req.path, req.body)
				if w.Code != 409 || !strings.Contains(w.Body.String(), tc.message) {
					t.Fatalf("%s %s status=%d body=%s want409/%s", req.method, req.path, w.Code, w.Body.String(), tc.message)
				}
			}
			// Ordinary approval does not require active stock dimensions.
			f.request(t, "admin", "301", "POST", fmt.Sprintf("/api/applications/%d/approve", app.ID), `{"auto_outbound":false}`, 200)
			companyGetPage[model.OutboundOrder](t, f, "beijing", 301, "/api/outbound", 0)
		})
	}
	var writes int
	if err := f.db.QueryRow(`SELECT (SELECT COUNT(*) FROM inventory_balances WHERE organization_id=301)+(SELECT COUNT(*) FROM inbound_orders WHERE organization_id=301)+(SELECT COUNT(*) FROM outbound_orders WHERE organization_id=301)+(SELECT COUNT(*) FROM stock_logs WHERE organization_id=301)`).Scan(&writes); err != nil || writes != 0 {
		t.Fatalf("configuration failures left writes=%d err=%v", writes, err)
	}
	f.request(t, "wuhan", "302", "POST", "/api/inbound", `{"handler":"h","supplier":"s","items":[{"material_id":302,"quantity":1}]}`, 201)
}

func TestCompanyIsolationApprovalConcurrencyAndFaultMySQL(t *testing.T) {
	f := newCompanyHTTPFixture(t)
	f.request(t, "both", "301", "POST", "/api/inbound", `{"handler":"h","supplier":"s","items":[{"material_id":301,"quantity":10}]}`, 201)
	f.request(t, "both", "302", "POST", "/api/inbound", `{"handler":"h","supplier":"s","items":[{"material_id":302,"quantity":20}]}`, 201)
	newApplication := func() int64 {
		t.Helper()
		var app model.MaterialApplication
		companyDecode(t, f.request(t, "both", "301", "POST", "/api/applications", `{"items":[{"material_id":301,"quantity":3}]}`, 201), &app)
		return app.ID
	}
	assertState := func(id int64, status, balance string, outbounds, logs int) {
		t.Helper()
		var gotStatus, gotBalance, otherBalance string
		var gotOrders, gotLogs int
		if err := f.db.QueryRow(`SELECT status FROM material_applications WHERE id=? AND organization_id=301`, id).Scan(&gotStatus); err != nil {
			t.Fatal(err)
		}
		if err := f.db.QueryRow(`SELECT on_hand FROM inventory_balances WHERE organization_id=301 AND material_id=301`).Scan(&gotBalance); err != nil {
			t.Fatal(err)
		}
		if err := f.db.QueryRow(`SELECT on_hand FROM inventory_balances WHERE organization_id=302 AND material_id=302`).Scan(&otherBalance); err != nil {
			t.Fatal(err)
		}
		if err := f.db.QueryRow(`SELECT COUNT(*) FROM outbound_orders WHERE organization_id=301`).Scan(&gotOrders); err != nil {
			t.Fatal(err)
		}
		if err := f.db.QueryRow(`SELECT COUNT(*) FROM stock_logs WHERE organization_id=301`).Scan(&gotLogs); err != nil {
			t.Fatal(err)
		}
		if gotStatus != status || gotBalance != balance || otherBalance != "20.000000" || gotOrders != outbounds || gotLogs != logs {
			t.Fatalf("state=%s/%s/%s/%d/%d want=%s/%s/20/%d/%d", gotStatus, gotBalance, otherBalance, gotOrders, gotLogs, status, balance, outbounds, logs)
		}
	}
	id := newApplication()
	start := make(chan struct{})
	results := make(chan *httptest.ResponseRecorder, 2)
	var wg sync.WaitGroup
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			<-start
			results <- f.serve("admin", "301", "POST", fmt.Sprintf("/api/applications/%d/approve", id), `{"auto_outbound":true}`)
		}()
	}
	close(start)
	wg.Wait()
	close(results)
	successes, conflicts := 0, 0
	for w := range results {
		switch w.Code {
		case 200:
			successes++
		case 400:
			conflicts++
		default:
			t.Fatalf("concurrent approval status=%d body=%s", w.Code, w.Body.String())
		}
	}
	if successes != 1 || conflicts != 1 {
		t.Fatalf("concurrent successes=%d conflicts=%d", successes, conflicts)
	}
	assertState(id, "approved", "7.000000", 1, 2)
	// Force failure after stock deduction and outbound insertion, at stock log
	// insertion. The transaction must restore all four artifacts.
	id = newApplication()
	companySQL(t, f.db, `CREATE TRIGGER fail_company_application_log BEFORE INSERT ON stock_logs FOR EACH ROW BEGIN IF NEW.organization_id=301 AND NEW.related_order_type='application' THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='integration injected log failure'; END IF; END`)
	f.request(t, "admin", "301", "POST", fmt.Sprintf("/api/applications/%d/approve", id), `{"auto_outbound":true}`, 500)
	assertState(id, "pending", "7.000000", 1, 2)
	companySQL(t, f.db, `DROP TRIGGER fail_company_application_log`)
	f.request(t, "admin", "301", "POST", fmt.Sprintf("/api/applications/%d/approve", id), `{"auto_outbound":true}`, 200)
	assertState(id, "approved", "4.000000", 2, 3)
}

func TestCompanyIsolationWebSocketRevocationMySQL(t *testing.T) {
	f := newCompanyHTTPFixture(t)
	withTicketService(t, 30*time.Second)
	f.router.GET("/ws", HandleWebSocket)
	server := httptest.NewServer(f.router)
	t.Cleanup(server.Close)
	ticket := func(account, org string) string {
		t.Helper()
		var result struct {
			Ticket string `json:"ticket"`
		}
		companyDecode(t, f.request(t, account, org, "POST", "/api/auth/websocket-ticket", "", 200), &result)
		if result.Ticket == "" {
			t.Fatal("empty WebSocket ticket")
		}
		return result.Ticket
	}
	type socketResult struct {
		event StockUpdateEvent
		err   error
	}
	connect := func(account, org string) <-chan socketResult {
		t.Helper()
		url := "ws" + strings.TrimPrefix(server.URL, "http") + "/ws?ticket=" + ticket(account, org)
		conn, response, err := websocket.DefaultDialer.Dial(url, nil)
		if err != nil {
			t.Fatalf("connect %s/%s: %v response=%v", account, org, err, response)
		}
		t.Cleanup(func() { conn.Close() })
		results := make(chan socketResult, 8)
		go func() {
			for {
				var event StockUpdateEvent
				err := conn.ReadJSON(&event)
				results <- socketResult{event, err}
				if err != nil {
					return
				}
			}
		}()
		return results
	}
	expectEvent := func(results <-chan socketResult, org int64, order int64) {
		t.Helper()
		select {
		case result := <-results:
			if result.err != nil || result.event.OrganizationID != org || result.event.OrderID != order {
				t.Fatalf("event=%+v want=%d/%d", result, org, order)
			}
		case <-time.After(3 * time.Second):
			t.Fatalf("no event for company %d", org)
		}
	}
	expectClosed := func(results <-chan socketResult) {
		t.Helper()
		select {
		case result := <-results:
			if result.err == nil {
				t.Fatalf("revoked socket received event=%+v", result.event)
			}
		case <-time.After(3 * time.Second):
			t.Fatal("revoked socket remained open")
		}
	}
	beijing := connect("both", "301")
	wuhan := connect("wuhan", "302")
	admin := connect("admin", "302")
	disabledCompany := connect("beijing", "301")
	// Both organizations are connected concurrently through real TCP handshakes.
	GetHub().BroadcastStockUpdate(301, "inbound", 701, nil)
	GetHub().BroadcastStockUpdate(302, "inbound", 702, nil)
	expectEvent(beijing, 301, 701)
	expectEvent(wuhan, 302, 702)
	expectEvent(admin, 302, 702)
	expectEvent(disabledCompany, 301, 701)
	for _, results := range []<-chan socketResult{beijing, wuhan, admin, disabledCompany} {
		select {
		case extra := <-results:
			t.Fatalf("cross-company/extra event=%+v", extra)
		case <-time.After(50 * time.Millisecond):
		}
	}
	// Recheck the database at delivery, rather than trusting handshake grants.
	beforeGrantRevocation := ticket("both", "301")
	companySQL(t, f.db, `DELETE FROM user_organizations WHERE user_id=303 AND organization_id=301`)
	f.request(t, "", "", "GET", "/ws?ticket="+beforeGrantRevocation, "", 403)
	GetHub().BroadcastStockUpdate(301, "outbound", 703, nil)
	expectClosed(beijing)
	expectEvent(disabledCompany, 301, 703)
	f.request(t, "both", "301", "GET", "/api/materials", "", 403)
	beforeRoleRevocation := ticket("admin", "302")
	companySQL(t, f.db, `DELETE FROM user_roles WHERE user_id=304; INSERT INTO user_roles(user_id,role_id) SELECT 304,id FROM roles WHERE code='warehouse_admin'`)
	GetHub().BroadcastStockUpdate(302, "outbound", 704, nil)
	expectClosed(admin)
	expectEvent(wuhan, 302, 704)
	f.request(t, "", "", "GET", "/ws?ticket="+beforeRoleRevocation, "", 403)
	f.request(t, "admin", "302", "GET", "/api/materials", "", 403)
	companySQL(t, f.db, `UPDATE organizations SET status='inactive' WHERE id=301`)
	GetHub().BroadcastStockUpdate(301, "outbound", 706, nil)
	expectClosed(disabledCompany)
	f.request(t, "beijing", "301", "GET", "/api/materials", "", 403)
	// A ticket issued before logout cannot start a new connection, and the live
	// socket cannot receive the next event. The old token also fails HTTP auth.
	pending := ticket("wuhan", "302")
	f.request(t, "wuhan", "", "POST", "/api/auth/logout", "", 200)
	f.request(t, "wuhan", "302", "GET", "/api/materials", "", 401)
	f.request(t, "wuhan", "302", "POST", "/api/auth/websocket-ticket", "", 401)
	f.request(t, "", "", "GET", "/ws?ticket="+pending, "", 401)
	GetHub().BroadcastStockUpdate(302, "outbound", 705, nil)
	expectClosed(wuhan)
	var revoked int
	if err := f.db.QueryRow(`SELECT COUNT(*) FROM revoked_tokens WHERE user_id=302`).Scan(&revoked); err != nil || revoked != 1 {
		t.Fatalf("logout persisted revocations=%d err=%v", revoked, err)
	}
	oldToken := f.tokens["wuhan"]
	var login struct {
		Token string `json:"token"`
	}
	companyDecode(t, f.request(t, "", "", "POST", "/api/auth/login", `{"username":"http-wuhan","password":"CompanyTest123!"}`, 200), &login)
	if login.Token == "" || login.Token == oldToken {
		t.Fatal("new login did not create a distinct session")
	}
	f.tokens["wuhan"] = login.Token
	f.request(t, "wuhan", "302", "GET", "/api/materials", "", 200)
}