import 'dart:async';

import 'package:async_tools/async_tools.dart';
import 'package:bloc/bloc.dart';
import 'package:bloc_tools/bloc_tools.dart';
import 'package:fast_immutable_collections/fast_immutable_collections.dart';
import 'package:mutex/mutex.dart';

import '../../veilid_support.dart';

typedef DHTShortArrayState<T> = AsyncValue<IList<T>>;
typedef DHTShortArrayBusyState<T> = BlocBusyState<DHTShortArrayState<T>>;

class DHTShortArrayCubit<T> extends Cubit<DHTShortArrayBusyState<T>>
    with BlocBusyWrapper<DHTShortArrayState<T>> {
  DHTShortArrayCubit({
    required Future<DHTShortArray> Function() open,
    required T Function(List<int> data) decodeElement,
  })  : _decodeElement = decodeElement,
        super(const BlocBusyState(AsyncValue.loading())) {
    Future.delayed(Duration.zero, () async {
      // Open DHT record
      _shortArray = await open();
      _wantsCloseRecord = true;

      // Make initial state update
      _update();
      _subscription = await _shortArray.listen(_update);
    });
  }

  DHTShortArrayCubit.value({
    required DHTShortArray shortArray,
    required T Function(List<int> data) decodeElement,
  })  : _shortArray = shortArray,
        _decodeElement = decodeElement,
        super(const BlocBusyState(AsyncValue.loading())) {
    // Make initial state update
    _update();
    Future.delayed(Duration.zero, () async {
      _subscription = await shortArray.listen(_update);
    });
  }

  Future<void> refresh({bool forceRefresh = false}) async => busy((emit) async {
        var out = IList<T>();
        // xxx could be parallelized but we need to watch out for rate limits
        for (var i = 0; i < _shortArray.length; i++) {
          final cir = await _shortArray.getItem(i, forceRefresh: forceRefresh);
          if (cir == null) {
            throw Exception('Failed to get short array element');
          }
          out = out.add(_decodeElement(cir));
        }
        emit(AsyncValue.data(out));
      });

  void _update() {
    // Run at most one background update process
    // Because this is async, we could get an update while we're
    // still processing the last one
    _sspUpdate.busyUpdate<T, AsyncValue<IList<T>>>(busy, (emit) async {
      try {
        final initialState = await _getElementsInner();
        emit(AsyncValue.data(initialState));
      } on Exception catch (e) {
        emit(AsyncValue.error(e));
      }
    });
  }

  // Get and decode the entire short array
  Future<IList<T>> _getElementsInner() async {
    assert(isBusy, 'should only be called from a busy state');
    var out = IList<T>();
    for (var i = 0; i < _shortArray.length; i++) {
      // Get the element bytes (throw if fails, array state is invalid)
      final bytes = (await _shortArray.getItem(i))!;
      // Decode the element
      final elem = _decodeElement(bytes);
      // Append to the output list
      out = out.add(elem);
    }
    return out;
  }

  @override
  Future<void> close() async {
    await _subscription?.cancel();
    _subscription = null;
    if (_wantsCloseRecord) {
      await _shortArray.close();
    }
    await super.close();
  }

  Future<R> operate<R>(Future<R> Function(DHTShortArray) closure) async =>
      _operateMutex.protect(() async => closure(_shortArray));

  final _operateMutex = Mutex();
  late final DHTShortArray _shortArray;
  final T Function(List<int> data) _decodeElement;
  StreamSubscription<void>? _subscription;
  bool _wantsCloseRecord = false;
  final _sspUpdate = SingleStatelessProcessor();
}
