<commit_before>/*
 * Copyright 2019 Google LLC.
 *
 * This code may only be used under the BSD style license found at
 * http://polymer.github.io/LICENSE.txt
 *
 * Code distributed by Google as part of this project is also subject to an additional IP rights
 * grant found at
 * http://polymer.github.io/PATENTS.txt
 */

package arcs.core.storage

import arcs.core.crdt.CrdtChange
import arcs.core.crdt.CrdtData
import arcs.core.crdt.CrdtException
import arcs.core.crdt.CrdtModel
import arcs.core.crdt.CrdtModelType
import arcs.core.crdt.CrdtOperation
import arcs.core.storage.DirectStore.State.Name.AwaitingDriverModel
import arcs.core.storage.DirectStore.State.Name.AwaitingResponse
import arcs.core.storage.DirectStore.State.Name.Idle
import arcs.core.storage.util.randomCallbackManager
import arcs.core.util.Random
import arcs.core.util.TaggedLog
import kotlin.coroutines.coroutineContext
import kotlinx.atomicfu.AtomicRef
import kotlinx.atomicfu.atomic
import kotlinx.atomicfu.getAndUpdate
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.ConflatedBroadcastChannel
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.launch

// import kotlinx.coroutines.flow.debounce
// import kotlinx.coroutines.flow.filter
// import kotlinx.coroutines.flow.first

/**
 * An [ActiveStore] capable of communicating directly with a [Driver].
 *
 * This is what *directly* manages a [CrdtSingleton], [CrdtSet], or [CrdtCount].
 */
@Suppress("EXPERIMENTAL_API_USAGE")
class DirectStore<Data : CrdtData, Op : CrdtOperation, T> /* internal */ constructor(
    options: StoreOptions,
    /* internal */
    val localModel: CrdtModel<Data, Op, T>,
    /* internal */
    val driver: Driver<Data>
) : ActiveStore<Data, Op, T>(options),
    WriteBack by StoreWriteBack.create(driver.storageKey.protocol) {
    override val versionToken: String?
        get() = driver.token

    // TODO(#5551): Consider including a hash of state.value and storage key in log prefix.
    private val log = TaggedLog { "DirectStore" }

    /** True if this store has been closed. */
    var closed = false

    /**
     * [AtomicRef] of a [CompletableDeferred] which will be completed when the [DirectStore]
     * transitions into the Idle state.
     */
    private val idleDeferred: AtomicRef<IdleDeferred> = atomic(CompletableDeferred(Unit))

    /**
     * [AtomicRef] of a list of [PendingDriverModel]s, allowing us to treat it as a copy-on-write,
     * threadsafe list using [AtomicRef.update].
     */
    private var pendingDriverModels = atomic(listOf<PendingDriverModel<Data>>())
    private var version = atomic(0)
    private var state: AtomicRef<State<Data>> = atomic(
        State.Idle(idleDeferred, driver)
    )
    private val stateChannel =
        ConflatedBroadcastChannel<State<Data>>(State.Idle(idleDeferred, driver))
    private val stateFlow = stateChannel.asFlow()
    private val callbackManager = randomCallbackManager<ProxyMessage<Data, Op, T>>(
        "direct",
        Random
    )
    private val scope = options.coroutineScope

    private val storeIdlenessFlow =
        combine(stateFlow, writebackIdlenessFlow) { state, writebackIsIdle ->
            state is State.Idle<*> && writebackIsIdle
        }

    override suspend fun idle() {
        /**
         * TODO: tune the debounce window
         * Once this is enabled, change [DirectStoreMuxer.idle] accordingly to use
         * simultaneous launches.
         */
        // storeIdlenessFlow.debounce(50).filter { it }.first()
    }

    fun getLocalData(): Data = synchronized(this) { localModel.data }

    override suspend fun on(callback: ProxyCallback<Data, Op, T>): Int {
        val callbackInvoke = callback::invoke
        synchronized(callbackManager) {
            return callbackManager.register(callbackInvoke)
        }
    }

    override fun off(callbackToken: Int) {
        synchronized(callbackManager) {
            callbackManager.unregister(callbackToken)
        }
    }

    /** Closes the store. Once closed, it cannot be re-opened. A new instance must be created. */
    override fun close() {
        synchronized(callbackManager) {
            callbackManager.callbacks.clear()
            closeInternal()
        }
    }

    private fun closeInternal() {
        if (!closed) {
            closed = true
            stateChannel.offer(State.Closed())
            stateChannel.close()
            state.value = State.Closed()
            closeWriteBack()
            requireNotNull(scope) {
                "store driver cannot be closed properly due to missing coroutine scope"
            }.run {
                launch { driver.close() }
            }
        }
    }

    /**
     * Receives operations/model-updates from connected storage proxies.
     *
     * Additionally, StorageProxy objects may request a SyncRequest, which will result in an
     * up-to-date model being sent back to that StorageProxy. A return value of `true` implies that
     * the message was accepted, a return value of `false` requires that the proxy send a model
     * sync.
     */
    override suspend fun onProxyMessage(
        message: ProxyMessage<Data, Op, T>
    ): Boolean {
        return when (message) {<commit_msg>suppress the exception due to missing scope (#6072)<commit_after>/*
 * Copyright 2019 Google LLC.
 *
 * This code may only be used under the BSD style license found at
 * http://polymer.github.io/LICENSE.txt
 *
 * Code distributed by Google as part of this project is also subject to an additional IP rights
 * grant found at
 * http://polymer.github.io/PATENTS.txt
 */

package arcs.core.storage

import arcs.core.crdt.CrdtChange
import arcs.core.crdt.CrdtData
import arcs.core.crdt.CrdtException
import arcs.core.crdt.CrdtModel
import arcs.core.crdt.CrdtModelType
import arcs.core.crdt.CrdtOperation
import arcs.core.storage.DirectStore.State.Name.AwaitingDriverModel
import arcs.core.storage.DirectStore.State.Name.AwaitingResponse
import arcs.core.storage.DirectStore.State.Name.Idle
import arcs.core.storage.util.randomCallbackManager
import arcs.core.util.Random
import arcs.core.util.TaggedLog
import kotlin.coroutines.coroutineContext
import kotlinx.atomicfu.AtomicRef
import kotlinx.atomicfu.atomic
import kotlinx.atomicfu.getAndUpdate
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.ConflatedBroadcastChannel
import kotlinx.coroutines.flow.asFlow
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.launch

// import kotlinx.coroutines.flow.debounce
// import kotlinx.coroutines.flow.filter
// import kotlinx.coroutines.flow.first

/**
 * An [ActiveStore] capable of communicating directly with a [Driver].
 *
 * This is what *directly* manages a [CrdtSingleton], [CrdtSet], or [CrdtCount].
 */
@Suppress("EXPERIMENTAL_API_USAGE")
class DirectStore<Data : CrdtData, Op : CrdtOperation, T> /* internal */ constructor(
    options: StoreOptions,
    /* internal */
    val localModel: CrdtModel<Data, Op, T>,
    /* internal */
    val driver: Driver<Data>
) : ActiveStore<Data, Op, T>(options),
    WriteBack by StoreWriteBack.create(driver.storageKey.protocol) {
    override val versionToken: String?
        get() = driver.token

    // TODO(#5551): Consider including a hash of state.value and storage key in log prefix.
    private val log = TaggedLog { "DirectStore" }

    /** True if this store has been closed. */
    var closed = false

    /**
     * [AtomicRef] of a [CompletableDeferred] which will be completed when the [DirectStore]
     * transitions into the Idle state.
     */
    private val idleDeferred: AtomicRef<IdleDeferred> = atomic(CompletableDeferred(Unit))

    /**
     * [AtomicRef] of a list of [PendingDriverModel]s, allowing us to treat it as a copy-on-write,
     * threadsafe list using [AtomicRef.update].
     */
    private var pendingDriverModels = atomic(listOf<PendingDriverModel<Data>>())
    private var version = atomic(0)
    private var state: AtomicRef<State<Data>> = atomic(
        State.Idle(idleDeferred, driver)
    )
    private val stateChannel =
        ConflatedBroadcastChannel<State<Data>>(State.Idle(idleDeferred, driver))
    private val stateFlow = stateChannel.asFlow()
    private val callbackManager = randomCallbackManager<ProxyMessage<Data, Op, T>>(
        "direct",
        Random
    )
    private val scope = options.coroutineScope

    private val storeIdlenessFlow =
        combine(stateFlow, writebackIdlenessFlow) { state, writebackIsIdle ->
            state is State.Idle<*> && writebackIsIdle
        }

    override suspend fun idle() {
        /**
         * TODO: tune the debounce window
         * Once this is enabled, change [DirectStoreMuxer.idle] accordingly to use
         * simultaneous launches.
         */
        // storeIdlenessFlow.debounce(50).filter { it }.first()
    }

    fun getLocalData(): Data = synchronized(this) { localModel.data }

    override suspend fun on(callback: ProxyCallback<Data, Op, T>): Int {
        val callbackInvoke = callback::invoke
        synchronized(callbackManager) {
            return callbackManager.register(callbackInvoke)
        }
    }

    override fun off(callbackToken: Int) {
        synchronized(callbackManager) {
            callbackManager.unregister(callbackToken)
        }
    }

    /** Closes the store. Once closed, it cannot be re-opened. A new instance must be created. */
    override fun close() {
        synchronized(callbackManager) {
            callbackManager.callbacks.clear()
            closeInternal()
        }
    }

    private fun closeInternal() {
        if (!closed) {
            closed = true
            stateChannel.offer(State.Closed())
            stateChannel.close()
            state.value = State.Closed()
            closeWriteBack()

            /**
             * The [scope] was initialized and assigned at the [StorageService.onBind], hence
             * the [driver.close] would only take effect in real use-cases / integration tests
             * but unit tests which don't spin up entire storage service stack.
             */
            scope?.launch { driver.close() }
        }
    }

    /**
     * Receives operations/model-updates from connected storage proxies.
     *
     * Additionally, StorageProxy objects may request a SyncRequest, which will result in an
     * up-to-date model being sent back to that StorageProxy. A return value of `true` implies that
     * the message was accepted, a return value of `false` requires that the proxy send a model
     * sync.
     */
    override suspend fun onProxyMessage(
        message: ProxyMessage<Data, Op, T>
    ): Boolean {
        return when (message) {
{'fraction_non_alphanumeric': 0.09329634766527971, 'fraction_numerical': 0.002404068423485899, 'mean_word_length': 2.9970436067997044, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2216968', 'n_tokens_mistral': 3159, 'n_tokens_neox': 3097, 'n_words': 861}