from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch

import pyarrow.parquet as pq

from solar_relay.cli import next_poll_at
from solar_relay.storage import Store, StoreError
from tests.test_status import observed


class ParquetTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.path = Path(self.directory.name) / "archive"
        self.store = Store(self.path)
        self.now = datetime(2026, 11, 1, 12, tzinfo=timezone.utc)

    def test_real_parquet_has_typed_readings_and_original_payload(self):
        value = observed(self.now)
        value["readings"]["boiler_enabled"] = False
        self.store.add(value)
        path = next(self.path.glob("date=*/samples.parquet"))
        self.assertEqual(path.read_bytes()[:4], b"PAR1")
        row = pq.ParquetFile(path).read().to_pylist()[0]
        self.assertEqual(row["production_ac_w"], 2000)
        self.assertIs(row["boiler_enabled"], False)
        self.assertIsNone(row["uploaded_at"])
        self.assertEqual(json.loads(row["payload"]), value)
        self.assertEqual(self.store.status()["format"], "parquet")

    def test_retention_uses_ack_time_and_never_deletes_pending(self):
        items = [observed(self.now - timedelta(days=90, minutes=index)) for index in range(4)]
        for item in items:
            self.store.add(item)
        self.store.acknowledge([items[0]["id"]], now=self.now - timedelta(days=31))
        self.store.acknowledge([items[1]["id"]], now=self.now - timedelta(days=30))
        self.store.acknowledge([items[2]["id"]], now=self.now)
        self.assertEqual(self.store.prune(now=self.now), 1)
        self.assertEqual({s["id"] for s in self.store.records()}, {s["id"] for s in items[1:]})
        self.assertEqual(self.store.pending(100), [items[3]])
        self.assertEqual(self.store.prune(now=self.now + timedelta(days=31)), 2)
        self.assertEqual(list(self.store.records()), [items[3]])

    def test_replay_cancels_expiry_until_the_next_acknowledgement(self):
        value = observed(self.now)
        self.store.add(value)
        self.store.acknowledge([value["id"]], now=self.now - timedelta(days=40))
        self.store.replay()
        self.assertEqual(self.store.prune(now=self.now), 0)
        self.store.acknowledge([value["id"]], now=self.now)
        self.assertEqual(self.store.prune(now=self.now), 0)

    def test_interrupted_atomic_write_preserves_the_previous_partition(self):
        first, second = observed(self.now), observed(self.now + timedelta(minutes=5))
        self.store.add(first)
        with patch("solar_relay.storage.os.replace", side_effect=OSError("simulated write failure")):
            with self.assertRaises(OSError):
                self.store.add(second)
        self.assertEqual(list(Store(self.path).records()), [first])
        self.assertEqual(list(self.path.glob("date=*/.samples-*.tmp")), [])

    def test_two_store_instances_cannot_lose_concurrent_records(self):
        other = Store(self.path)
        values = [observed(self.now + timedelta(minutes=index * 5)) for index in range(12)]
        def add(pair):
            index, value = pair
            (other if index % 2 else self.store).add(value)
        with ThreadPoolExecutor(max_workers=2) as workers:
            list(workers.map(add, enumerate(values)))
        self.assertEqual({value["id"] for value in self.store.records()}, {value["id"] for value in values})

    def test_pruning_an_entire_day_removes_its_parquet_file(self):
        value = observed(self.now - timedelta(days=60))
        self.store.add(value)
        self.store.acknowledge([value["id"]], now=self.now - timedelta(days=31))
        self.assertEqual(self.store.prune(now=self.now), 1)
        self.assertEqual(list(self.path.glob("date=*/samples.parquet")), [])
        self.assertEqual(self.store.status()["samples"], 0)

    def test_corrupt_archive_fails_without_silently_dropping_records(self):
        value = observed(self.now)
        self.store.add(value)
        next(self.path.glob("date=*/samples.parquet")).write_bytes(b"broken")
        with self.assertRaises(StoreError):
            self.store.pending(100)

    def test_archive_path_requires_a_directory(self):
        occupied = self.path.parent / "existing-file"
        occupied.write_bytes(b"existing data")
        with self.assertRaises(StoreError):
            Store(occupied)
        self.assertEqual(occupied.read_bytes(), b"existing data")

    def test_next_poll_is_aligned_to_the_five_minute_clock(self):
        self.assertEqual(next_poll_at(301, 300), 600)
        self.assertEqual(next_poll_at(599.5, 300), 600)
        self.assertEqual(next_poll_at(600, 300), 900)


if __name__ == "__main__":
    unittest.main()