-- Copyright (c) 2025 Huawei Technologies Co., Ltd.
-- openUBMC is licensed under Mulan PSL v2.
-- You can use this software according to the terms and conditions of the Mulan PSL v2.
-- You may obtain a copy of Mulan PSL v2 at: http://license.coscl.org.cn/MulanPSL2
-- THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
-- EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
-- MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
-- See the Mulan PSL v2 for more details.

local skynet = require 'skynet'
local log = require 'mc.logging'
local context = require 'mc.context'
local vos = require 'utils.vos'
local ipmi = require 'ipmi'
local comp_code = ipmi.types.Cc
local bs = require "mc.bitstring"
local chip_lock_singleton = require 'chip_lock'
local msg = require 'general_hardware.ipmi.ipmi_message'
local defs = require 'unit_manager.class.logic_fw.comm_defs'
local drivers = require 'unit_manager.class.logic_fw.upgrade.drivers_api'
local chip = require 'unit_manager.class.logic_fw.cpld_chip'
local client = require 'general_hardware.client'
local fructl = require 'mcu.upgrade.fructl_handler'
local gen_hw_bus = require 'general_hardware_bus'
local gpio = require 'libsoc_adapter.gpio'

local cpld_space_test = {}

cpld_space_test.STATUS = {
    SUCCESS = 0,
    FAIL = 1,
    NOTSUPPORT = 2,
    TESTING = 3
}


local test_support_list = {
    -- 安路
    anlu_ef2l15 = { id = 0x04004c37, manufacturer_id = 0 },
    anlu_ef2 = { id = 0x03004c37, manufacturer_id = 0 },
    anlu_ef3 = { id = 0x00008c3b, manufacturer_id = 0 },
    anlu_3f0f9 = { id = 0x0003f0f9, manufacturer_id = 0 },
    -- lattice
    lattice_ef2 = { id = 0x012bd043, manufacturer_id = 1 },
    lattice_ef3 = { id = 0x612bd043, manufacturer_id = 1 },
    lattice_lcmxo2_4000hc = { id = 0x012bc043, manufacturer_id = 1 },
    -- 紫光
    pango_ef2 = { id = 0x0042a899, manufacturer_id = 2 },
    pango_ef3 = { id = 0x0042b988, manufacturer_id = 2 },
    pango = { id = 0x0042b899, manufacturer_id = 2 },
}

local function get_space_test_obj(chip_index, chip_id, test_obj_list)
    for _, test_obj in ipairs(test_obj_list) do
        if test_obj.CpldIndex ~= chip_index then
            goto continue
        end
        for _, item in pairs(test_support_list) do
            if chip_id == item.id and test_obj.Manufacturer == item.manufacturer_id then
                log:notice("Test obj %s need check or recover cpld.", test_obj.name)
                return test_obj
            end
        end
        ::continue::
    end
    return nil
end

function cpld_space_test.start_cpld_test(req, ctx)
    local test_hpm_name = req.FirmwareName
    if string.len(test_hpm_name) ~= req.FirmwareFileLength then
        log:error('Test cpld hpm %s name length is out of range.', test_hpm_name)
        ipmi.ipmi_operation_log(ctx, 'general_hardware', 'Start cpld space test fail.')
        return msg.StartCpldSpaceTestRsp.new(comp_code.InvalidFieldRequest)
    end

    if not vos.get_file_accessible('/tmp/' .. test_hpm_name) then
        log:error('Test cpld hpm %s not exist', test_hpm_name)
        ipmi.ipmi_operation_log(ctx, 'general_hardware', 'Start cpld space test fail.')
        return msg.StartCpldSpaceTestRsp.new(comp_code.UnspecifiedError)
    end

    local ok, ret = client:PUpdateServiceUpdateServiceStartUpgrade(context.new(), '/tmp/' .. test_hpm_name, {})
    if not ok then
        log:error('start test hpm upgrade task failed, msg = %s', ret)
        ipmi.ipmi_operation_log(ctx, 'general_hardware', 'Start cpld space test fail.')
        return msg.StartCpldSpaceTestRsp.new(comp_code.UnspecifiedError)
    end
    ipmi.ipmi_operation_log(ctx, 'general_hardware', 'Start cpld space test successfully.')
    return msg.StartCpldSpaceTestRsp.new(comp_code.Success)
end

local function get_cpld_check_result(logic_fw, test_obj, chip_index, case_index)
    local ok, result
    if test_obj and test_obj.RefMode == 'GPIO' then
        local gpio_drv = gpio.new()
        ok, result = pcall(gpio.init, gpio_drv, test_obj.RefBusIndex, 0)
        if ok then
            _, result = pcall(gpio.read, gpio_drv, test_obj.RefBusIndex, 0)
        end
        ok, _ = pcall(gpio.close, gpio_drv)
        if not ok then
            log:error('Close gpio_drv failed.')
        end
    else
        local retry = 0
        ok, result = pcall(function()
            return test_obj.CpldResultRegister
        end)
        while not ok do
            skynet.sleep(300)
            ok, result = pcall(function()
                return test_obj.CpldResultRegister
            end)
            retry = retry + 1
            if retry == 3 then
                break
            end
        end
    end
    local check_result = (ok and result == 0) and true or false
    log:maintenance(log.MLOG_INFO, log.FC__PUBLIC_OK, '%s cpld%s test case%s %s.',
        logic_fw.device_name, chip_index, case_index, (check_result and 'pass' or 'fail'))
    return check_result
end


-- 加载cpld,将升级文件拷贝到cpld flash中
local function load_cpld_single(obj_fw, upg_file, file_type)
    local CPLD_UPGRADE_LOAD_MODE = 0
    local file_name_len = #upg_file
    local MAX_CPLD_LEN = 0x300000 -- 驱动限制为3M
    local file_len = vos.get_file_length(upg_file)
    if (file_len > MAX_CPLD_LEN or file_len <= 0) and file_type ~= 'svf' then
        log:error('file %s len:0x%x(Max:0x%x) invalid!', upg_file, file_len, MAX_CPLD_LEN)
        return defs.RET.ERR
    end
    log:notice('load file[0x%x]', file_len)
    local upg_info = {
        mode = CPLD_UPGRADE_LOAD_MODE,
        product_id = 0,
        board_id = 0,
        pcb = 0x43,       -- V2中用字符C赋值，这里直接用ascii码0x43表示
        component_id = 0, -- V3采用CSR配置每一种JTAG链路模式，不用再进行链路选择，cid默认填0
        file_type = file_type,
        filename_len = file_name_len,
        filename = upg_file
    }
    -- 按照V2 CPLD_UPG_FILE_S结构体小端对齐封装
    -- V2的CPLD_UPG_FILE_S结构体设计未考虑字节序对齐，mode需要修改为4字节
    local CPLD_UPG_FILE_DESC = string.format(
        '<<mode:4/unit:8, product_id:4/little-unit:8, ' ..
        'board_id:4/little-unit:8, ' ..
        'pcb:4/little-unit:8, component_id:4/little-unit:8, ' ..
        'file_type:3/string, filename_len:4/little-unit:8, filename:%d/binary>>',
        file_name_len)
    local info = bs.new(CPLD_UPG_FILE_DESC):pack(upg_info)
    local ok, error = pcall(obj_fw.update_chip.Write, obj_fw.update_chip,
        require 'mc.context'.new(), 0, info)
    if not ok then
        log:error('[cpld]block_write fail:%s', error)
        return defs.RET.ERR
    end
    log:notice('[cpld]%s update success!', upg_file)
    return defs.RET.OK
end

-- 将CPLD检测用的case文件加载到SRAM中， 每个case最多重试3次
local function load_test_case_single(logic_fw, cpld_case_path, file_type)
    for count = 1, 3 do
        local ret = load_cpld_single(logic_fw, cpld_case_path, file_type)
        if ret == defs.RET.OK then
            return ret
        end
        log:error('write %s case to sram failed, fail times = %s', cpld_case_path, count)
        --每次写入失败，等30s再重试
        skynet.sleep(3000)
    end
    return defs.RET.ERR
end

local function load_test_cpld_process(logic_fw, test_obj, file_path, file_type)
    local index = test_obj.CpldIndex
    drivers:set_upg_cpld_chip(logic_fw.chip_info, index - 1)
    local ret
    local check_flag = true
    local check_result
    for k, v in ipairs(test_obj.TestFirmwares) do
        local cpld_case_path = string.format('%s%s', file_path, v)
        ret = load_test_case_single(logic_fw, cpld_case_path, file_type)
        if ret == defs.RET.OK then
            skynet.sleep(test_obj.WaitTime / 10)
            check_result = get_cpld_check_result(logic_fw, test_obj, index, k)
        else
            check_result = false
            log:maintenance(log.MLOG_INFO, log.FC__PUBLIC_OK, '%s cpld%s test case%s fail after write error.',
                logic_fw.device_name, index, k)
        end
        if not check_result then
            check_flag = false
        end
    end

    --所有case都测试通过才被认为检测成功
    local result = check_flag and
        cpld_space_test.STATUS.SUCCESS or cpld_space_test.STATUS.FAIL
    cpld_space_test.save_cpld_test_result(logic_fw, index, result)
end

local function recover_cpld_process(logic_fw, test_obj, file_path, file_type)
    if test_obj.CpldIndex < 1 then
        log:error('Get test_obj CpldIndex error.')
        return defs.RET.ERR
    end
    --CPLD 的chip序号从0开始
    drivers:set_upg_cpld_chip(logic_fw.chip_info, test_obj.CpldIndex - 1)
    local recover_path = string.format('%s%s', file_path, test_obj.RefreshFirmware)
    local ret = load_cpld_single(logic_fw, recover_path, file_type)
    local result = (ret == defs.RET.OK and 'successfully' or 'failed')
    log:notice('update %s %s.', recover_path, result)
    log:maintenance(log.MLOG_INFO, log.FC__PUBLIC_OK, '%s cpld%s recover %s.',
        logic_fw.device_name, test_obj.CpldIndex, result)
end

local function action_for_cpld_test(cfg, release_path, fw_list, action_type)
    if not cfg[action_type] then
        return defs.RET.OK
    end

    local toolboard_logicfws = {}
    for _, v in ipairs(fw_list) do
        for uid in string.gmatch(cfg.uid_action, "%w+") do
            if v.uid == uid then
                table.insert(toolboard_logicfws, v)
            end
        end
    end
    if #toolboard_logicfws == 0 then
        log:error('Find toolboard logicfw failed')
        return defs.RET.ERR
    end

    for _, toolboard_logicfw in ipairs(toolboard_logicfws) do
        local cpld_info = {}
        if drivers:get_cpld_device_info(toolboard_logicfw.chip_info, cpld_info) ~= defs.RET.OK then
            log:error('Get %s cpld device id failed.', toolboard_logicfw.device_name)
            return defs.RET.ERR
        end

        toolboard_logicfw:switch_to_firmware_route()
        local ok, _ = pcall(toolboard_logicfw.chip_info.SetBypassMode, toolboard_logicfw.chip_info, context.new(), true)
        if not ok then
            log:error('Set %s cpld bypass mode failed.', toolboard_logicfw.device_name)
            toolboard_logicfw:switch_to_default_route()
            return defs.RET.ERR
        end
        local file_id = chip:get_cpld_upgrade_file_id(cpld_info['device1_id'], 1, 1)
        local action_file_path = string.format('%s%s%02d.%s', release_path, cfg[action_type], file_id, cfg.file_type)
        local ret = load_cpld_single(toolboard_logicfw, action_file_path, cfg.file_type)
        toolboard_logicfw:switch_to_default_route()

        log:maintenance(log.MLOG_INFO, log.FC__PUBLIC_OK, '%s write %s cpld %s.',
            toolboard_logicfw.device_name, cfg[action_type], (ret == defs.RET.OK and 'successfully' or 'failed'))
        if ret ~= defs.RET.OK then
            log:error('%s write %s file failed.', toolboard_logicfw.device_name, action_type)
            return ret
        end
        skynet.sleep(1000)
    end
    return defs.RET.OK
end

local function load_test_cpld(logic_fw, cfg, release_path, fw_list)
    local cpld_info = {}
    if drivers:get_cpld_device_info(logic_fw.chip_info, cpld_info) ~= defs.RET.OK then
        log:error('Get %s cpld device id failed.', logic_fw.device_name)
        cpld_space_test.save_cplds_test_result(logic_fw, logic_fw.test_chip_nums, cpld_space_test.STATUS.FAIL)
        return defs.RET.OK
    end
    local ok, _ = pcall(logic_fw.chip_info.SetBypassMode, logic_fw.chip_info, context.new(), true)
    if not ok then
        log:error('set %s cpld bypass mode failed.', logic_fw.device_name)
        cpld_space_test.save_cplds_test_result(logic_fw, logic_fw.test_chip_nums, cpld_space_test.STATUS.FAIL)
        return defs.RET.OK
    end
    local power_states = fructl.get_power_status(gen_hw_bus.get_instance().bus, 1)
    if power_states ~= 'OFF' then
        log:error('current power state %s not support test.', power_states)
        cpld_space_test.save_cplds_test_result(logic_fw, logic_fw.test_chip_nums, cpld_space_test.STATUS.FAIL)
        return defs.RET.OK
    end

    if action_for_cpld_test(cfg, release_path, fw_list, 'before_action') ~= defs.RET.OK then
        cpld_space_test.save_cplds_test_result(logic_fw, logic_fw.test_chip_nums, cpld_space_test.STATUS.FAIL)
        return defs.RET.OK
    end
    local chip_id, test_obj
    for i = 1, cpld_info.device_count do
        chip_id = cpld_info[string.format('device%d_id', i)]
        test_obj = get_space_test_obj(i, chip_id, logic_fw.test_object)
        if not test_obj then
            log:maintenance(log.MLOG_INFO, log.FC__PUBLIC_OK, '%s cpld %s chipid 0x%x not need test.',
                logic_fw.device_name, i, chip_id)
            cpld_space_test.save_cpld_test_result(logic_fw, i, cpld_space_test.STATUS.NOTSUPPORT)
            goto continue
        end
        load_test_cpld_process(logic_fw, test_obj, release_path, cfg.file_type)

        ::continue::
    end

    for i = 1, cpld_info.device_count do
        chip_id = cpld_info[string.format('device%d_id', i)]
        test_obj = get_space_test_obj(i, chip_id, logic_fw.test_object)
        if not test_obj then
            log:maintenance(log.MLOG_INFO, log.FC__PUBLIC_OK, '%s cpld %s chipid 0x%x not need recover.',
                logic_fw.device_name, i, chip_id)
            goto continue
        end
        recover_cpld_process(logic_fw, test_obj, release_path, cfg.file_type)

        ::continue::
    end

    action_for_cpld_test(cfg, release_path, fw_list, 'after_action')
    return defs.RET.OK
end

local function set_smc_accessibility(logic_fw, smc_obj, accessibility, timeout)
    if not smc_obj then
        log:error('Set smc accessibility failed, smc_obj is nil')
        return
    end

    local ok, error_msg = pcall(smc_obj.SetAccessibility, smc_obj, context.new(), accessibility, timeout)
    if ok then
        log:notice('set %s smc chip accessibility %s successfully.', logic_fw.device_name, accessibility)
    else
        log:error('set %s smc chip accessibility %s failed, error: %s.', logic_fw.device_name, accessibility, error_msg)
    end
end

local function update_chip_lock(ctx, lock_time, logic_fw)
    return pcall(function(...)
        return chip_lock_singleton.get_instance():lock(logic_fw.update_lock_chip, ctx, lock_time)
    end)
end

local function update_chip_unlock(ctx, logic_fw)
    return pcall(function(...)
        return chip_lock_singleton.get_instance():unlock(logic_fw.update_lock_chip, ctx)
    end)
end

-- 基础板CPLD通过MCU上报结果，在自检前后需要给MCU设置相关信号
local function set_cpld_test_mode(logic_fw, smc_obj, flag)
    if not smc_obj then
        log:error('Set cpld test mode failed, smc_obj nil.')
        return
    end
    local ok, error_msg = pcall(smc_obj['bmc.kepler.Chip.BlockIO'].Write,
        smc_obj['bmc.kepler.Chip.BlockIO'], context.new(), 0x00000E00, flag)

    if ok then
        log:notice('Set %s cpld test mode to %s successfully.', logic_fw.device_name, flag)
    else
        log:error('Set %s cpld test mode to %s failed, error: %s.',
            logic_fw.device_name, flag, error_msg)
    end
end

function cpld_space_test.load_test_cpld_with_lock(logic_fw, cfg, release_path, fw_list)
    local lock_time = 600
    local ret
    if #logic_fw.test_object == 0 then
        log:error('get space test object list empty, fw id = %s.', logic_fw.id)
        cpld_space_test.save_cplds_test_result(logic_fw, logic_fw.test_chip_nums, cpld_space_test.STATUS.FAIL)
        return defs.RET.OK
    end

    local smc_obj
    local ref_mode_mcu = false
    for _, v in ipairs(logic_fw.test_object) do
        if v.RefSmcChip then
            smc_obj = v.RefSmcChip
            if v.RefMode == 'MCU' then
                ref_mode_mcu = true
            end
            break
        end
    end
    -- 从MCU获取自检结果，MCU会保证SMC无误告警
    if ref_mode_mcu then
        set_cpld_test_mode(logic_fw, logic_fw.smc_chip, string.char(0x02))
    else
        -- 停止CPLD芯片关联的SMC通道访问，超时时间1800s
        set_smc_accessibility(logic_fw, smc_obj, false, 1800)
    end
    while lock_time > 0 do
        local ok, ret_code = update_chip_lock(context.new(), 600, logic_fw)
        if ok and ret_code == defs.RET.OK then
            ret = load_test_cpld(logic_fw, cfg, release_path, fw_list)
            ok, ret_code = update_chip_unlock(context.new(), logic_fw)
            if not ok or ret_code ~= defs.RET.OK then
                log:error('unlock %s cpld chip failed, ret = %s.', logic_fw.device_name, ret_code)
            end
            set_smc_accessibility(logic_fw, smc_obj, true, 1)
            return ret
        end
        lock_time = lock_time - 1
        skynet.sleep(100)
    end

    if ref_mode_mcu then
        set_cpld_test_mode(logic_fw, logic_fw.smc_chip, string.char(0x00))
    else
        -- 停止CPLD芯片关联的SMC通道访问，超时时间1800s
        set_smc_accessibility(logic_fw, smc_obj, true, 1)
    end
    cpld_space_test.save_cplds_test_result(logic_fw, logic_fw.test_chip_nums, cpld_space_test.STATUS.FAIL)
    return defs.RET.OK
end

function cpld_space_test.get_cpld_test_result(req, ctx)
    local result = {}
    local db = cpld_space_test.app_db
    if not db then
        log:error('get db failed')
        return msg.GetCpldSpaceTestResultRsp.new(comp_code.DataNotAvailable, #result, table.concat(result))
    end

    local records = db:select(db.CpldSpaceTest):where(db.CpldSpaceTest.CheckResult:ne(255))
        :order_by(db.CpldSpaceTest.CpldPosition):all()

    for _, record in ipairs(records) do
        result[#result + 1] = string.char(record.CheckResult)
    end

    return msg.GetCpldSpaceTestResultRsp.new(comp_code.Success, #result, table.concat(result))
end

function cpld_space_test.init(db)
    cpld_space_test.app_db = db
    return cpld_space_test
end

function cpld_space_test.get_test_obj(logic_fw, position)
    for _, test_obj in pairs(logic_fw.test_object) do
        if test_obj.CpldPosition and test_obj.CpldPosition == position then
            return test_obj
        end
    end

    log:error('get test obj of position %s err.', position)
    return nil
end

function cpld_space_test.save_cpld_test_result(logic_fw, index, result)
    local cpld_name = string.match(logic_fw.csr.Name, "([^_]+)")
    local cpld_position = cpld_name .. '_' .. logic_fw.position .. '_Cpld' .. index
    local test_obj = cpld_space_test.get_test_obj(logic_fw, cpld_position)
    if test_obj then
        test_obj.CheckResult = result
        log:notice('update CPLD: %s, result: %s successfully', cpld_position, result)
        return
    end
    log:error('update CPLD: %s, result: %s failed', cpld_position, result)
end

function cpld_space_test.save_cplds_test_result(logic_fw, cpld_num, result)
    for index = 1, cpld_num do
        cpld_space_test.save_cpld_test_result(logic_fw, index, result)
    end
end

return cpld_space_test
