-- This Source Code Form is subject to the terms of the Mozilla Public
-- License, v. 2.0. If a copy of the MPL was not distributed with this
-- file, You can obtain one at http://mozilla.org/MPL/2.0/.

--[[
# Hindsight input/analysis log file pruner

Deletes the log files generated by the input and/or analysis plugins when all
the analysis and output plugins are done consumining them (within
ticker_interval seconds).

## Sample Configuration
```lua
filename = "prune_input.lua"
ticker_interval = 60
input = true
analysis = true

-- When true, causes the plugin to stop/abort when the checkpoints are no longer
-- advancing. Use this option to allow hindsight_cli to exit when the inputs are
-- finished. This option is typically used when streaming a large data set from
-- something like s3 i.e., running a report.
exit_on_stall = false
```
--]]
require "io"
require "os"
require "string"
require "math"
local l = require "lpeg"
l.locale(l)

local output_path   = read_config("output_path") -- provided by Hindsight
local exit_on_stall = read_config("exit_on_stall")
local input = read_config("input") and l.P"input"
local analysis = read_config("analysis") and l.P"analysis"
if not (input or analysis) then
    error("either input or analysis (or both) must be set")
end

local function get_min(t, n, i, o)
    local elt = t[n] or {min = math.huge}
    if i < elt.min then
        elt.min = i
        elt.off = o
    end
    t[n] = elt
    return t
end

local pair      = l.P"'" * l.Cg(l.digit^1/tonumber * ":" * l.C(l.digit^1)) * "'"
local ignore    = (l.P(1) - "\n")^0 * "\n"
local plugins
if input and analysis then
    plugins = input + analysis
else
    plugins = input or analysis
end
local line = l.P"_G['" * l.Cg(l.C(plugins) * "->" * (l.P(1) - "'")^1 * "']" * l.space^0 * "=" * l.space^0 * pair) * l.space^0 + ignore
local grammar = l.Cf(l.Ct("") * line^1, get_min)
local old_cp = {}


function process_message()
    local fh = io.open(output_path .. "/hindsight.cp")
    if not fh then return 0 end -- checkpoint file not available yet

    local s = fh:read("*a")
    fh:close()
    if s then
        local stalled = true
        local t = grammar:match(s)
        if t then
            for name, elt in pairs(t) do
                local cp = old_cp[name] or {}
                if stalled and cp.min ~= elt.min or cp.off ~= elt.off then
                    stalled = false
                end
                cp.off = elt.off
                if cp.min ~= elt.min then
                    cp.min = elt.min
                    for i = cp.min - 1, 0, -1 do
                        local r = os.remove(string.format("%s/%s/%d.log", output_path, name, i))
                        if not r then break end
                    end
                end
                old_cp[name] = cp
            end
            if stalled and exit_on_stall then
                error("input/analysis have stopped")
            end
        end
    end
    return 0
end
