// Copyright 2015-2021 Swim Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#[cfg(test)]
mod tests;

use crate::uplink::{WarpUplinkProfile, WarpUplinkPulse};
use crate::{MetricReporter, MetricStage};
use std::ops::Add;
use swim_form::Form;

/// An accumulated WARP lane profile generated by the accumulator task.
#[derive(Default, Form, Copy, Clone, PartialEq, Debug)]
pub struct WarpLaneProfile {
    // todo: WarpDownlinkProfile aggregation
    // todo: LaneAddress
    /// The number of events that have been produced since the last report time.
    pub uplink_event_delta: u32,
    /// The rate at which events were produced since the last report time.
    pub uplink_event_rate: u64,
    /// The number of command messages that have been produced since the last report time.
    pub uplink_command_delta: u32,
    /// The rate at which command messages were produced since the last report time.
    pub uplink_command_rate: u64,
    /// The number of uplinks that were opened since the last report time.
    pub uplink_open_delta: u32,
    /// The number of uplinks that were closed since the last report time.
    pub uplink_close_delta: u32,
}

impl Add<WarpLaneProfile> for WarpLaneProfile {
    type Output = WarpLaneProfile;

    fn add(self, rhs: WarpLaneProfile) -> Self::Output {
        WarpLaneProfile {
            uplink_event_delta: self
                .uplink_event_delta
                .saturating_add(rhs.uplink_event_delta),
            uplink_event_rate: self.uplink_event_rate.saturating_add(rhs.uplink_event_rate),
            uplink_command_delta: self
                .uplink_command_delta
                .saturating_add(rhs.uplink_command_delta),
            uplink_command_rate: self
                .uplink_command_rate
                .saturating_add(rhs.uplink_command_rate),
            uplink_open_delta: self.uplink_open_delta.saturating_add(rhs.uplink_open_delta),
            uplink_close_delta: self
                .uplink_close_delta
                .saturating_add(rhs.uplink_close_delta),
        }
    }
}

#[derive(Default)]
pub struct LaneMetricReporter {
    event_count: u64,
    command_count: u64,
    close_count: u32,
    open_count: u32,
}

impl MetricReporter for LaneMetricReporter {
    const METRIC_STAGE: MetricStage = MetricStage::Lane;

    type Pulse = LanePulse;
    type Profile = WarpLaneProfile;
    type Input = WarpUplinkProfile;

    fn report(&mut self, part: Self::Input) -> (Self::Pulse, Self::Profile) {
        let LaneMetricReporter {
            event_count,
            command_count,
            close_count,
            open_count,
        } = self;

        let WarpUplinkProfile {
            event_delta,
            event_rate,
            command_delta,
            command_rate,
            open_delta,
            close_delta,
        } = part;

        *open_count = open_count.saturating_add(open_delta);
        *close_count = close_count.saturating_add(close_delta);
        *event_count = event_count.saturating_add(event_delta as u64);
        *command_count = command_count.saturating_add(command_delta as u64);

        let link_count = open_count.saturating_sub(*close_count);

        let pulse = LanePulse {
            uplink_pulse: WarpUplinkPulse {
                link_count,
                event_rate,
                event_count: *event_count,
                command_rate,
                command_count: *command_count,
            },
        };

        let profile = WarpLaneProfile {
            uplink_event_delta: event_delta,
            uplink_event_rate: event_rate,
            uplink_command_delta: command_delta,
            uplink_command_rate: command_rate,
            uplink_open_delta: open_delta,
            uplink_close_delta: close_delta,
        };

        (pulse, profile)
    }
}

impl Add<WarpUplinkProfile> for WarpLaneProfile {
    type Output = WarpLaneProfile;

    fn add(self, rhs: WarpUplinkProfile) -> Self::Output {
        let WarpLaneProfile {
            uplink_event_delta,
            uplink_event_rate,
            uplink_command_delta,
            uplink_command_rate,
            uplink_open_delta,
            uplink_close_delta,
        } = self;

        let WarpUplinkProfile {
            event_delta,
            event_rate,
            command_delta,
            command_rate,
            open_delta,
            close_delta,
        } = rhs;

        WarpLaneProfile {
            uplink_event_delta: uplink_event_delta.saturating_add(event_delta),
            uplink_event_rate: uplink_event_rate.saturating_add(event_rate),
            uplink_command_delta: uplink_command_delta.saturating_add(command_delta),
            uplink_command_rate: uplink_command_rate.saturating_add(command_rate),
            uplink_open_delta: uplink_open_delta.saturating_add(open_delta),
            uplink_close_delta: uplink_close_delta.saturating_add(close_delta),
        }
    }
}

/// A lane pulse detailing accumulated metrics.
#[derive(Default, Form, Clone, PartialEq, Debug)]
pub struct LanePulse {
    /// Accumulated WARP uplink pulse.
    #[form(name = "uplinkPulse")]
    pub uplink_pulse: WarpUplinkPulse,
}

impl Add<LanePulse> for LanePulse {
    type Output = LanePulse;

    fn add(mut self, rhs: LanePulse) -> Self::Output {
        self.uplink_pulse = self.uplink_pulse.add(rhs.uplink_pulse);
        self
    }
}
