mirror of
https://github.com/nolen777/eagle0.git
synced 2026-07-28 21:35:42 +00:00
Store client text in Postgres for Postgres games (#6985)
This commit is contained in:
@@ -83,6 +83,31 @@ scala_library(
|
||||
],
|
||||
)
|
||||
|
||||
scala_library(
|
||||
name = "postgres_client_text_store",
|
||||
srcs = ["PostgresClientTextStore.scala"],
|
||||
visibility = [
|
||||
"//src/main/scala/net/eagle0/eagle:__subpackages__",
|
||||
"//src/test/scala/net/eagle0/eagle:__subpackages__",
|
||||
],
|
||||
runtime_deps = [
|
||||
"@maven//:org_postgresql_postgresql",
|
||||
],
|
||||
deps = [
|
||||
":client_text",
|
||||
":client_text_store",
|
||||
":pregenerated_text_store",
|
||||
":text_generation_result",
|
||||
"//src/main/protobuf/net/eagle0/eagle/internal:llm_request_scala_proto",
|
||||
"//src/main/scala/net/eagle0/common:jfr_events",
|
||||
"//src/main/scala/net/eagle0/eagle:eagle_pkg",
|
||||
"//src/main/scala/net/eagle0/eagle/library:eagle_internal_exception",
|
||||
"//src/main/scala/net/eagle0/eagle/library/util:eagle_require",
|
||||
"//src/main/scala/net/eagle0/eagle/model/action_result/generated_text_request",
|
||||
"//src/main/scala/net/eagle0/eagle/model/proto_converters:generated_text_request_converter",
|
||||
],
|
||||
)
|
||||
|
||||
scala_library(
|
||||
name = "sqlite_client_text_store",
|
||||
srcs = ["SqliteClientTextStore.scala"],
|
||||
|
||||
@@ -0,0 +1,828 @@
|
||||
package net.eagle0.eagle.client_text
|
||||
|
||||
import java.sql.{Connection, DriverManager, ResultSet}
|
||||
|
||||
import scala.util.Try
|
||||
|
||||
import net.eagle0.common.JfrEvents
|
||||
import net.eagle0.eagle.{ClientTextId, FactionId, GameId}
|
||||
import net.eagle0.eagle.library.util.EagleRequire.internalRequire
|
||||
import net.eagle0.eagle.library.EagleInternalException
|
||||
import net.eagle0.eagle.model.action_result.generated_text_request.GeneratedTextRequestT
|
||||
import net.eagle0.eagle.model.proto_converters.GeneratedTextRequestConverter
|
||||
|
||||
class PostgresClientTextStore private[client_text] (
|
||||
private[client_text] val connection: Connection,
|
||||
gameId: GameId,
|
||||
pregenerated: PregeneratedClientTextStore
|
||||
) extends ClientTextStore {
|
||||
|
||||
private val databaseName = s"postgres-text:${gameId.toHexString}"
|
||||
|
||||
override def completeTexts: Map[ClientTextId, CompleteClientText] = {
|
||||
val sql = "SELECT id, text, requested_after_history_count FROM client_texts WHERE status = 'complete'"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.completeTexts", sql) { event =>
|
||||
try {
|
||||
val rs = stmt.executeQuery()
|
||||
val result = scala.collection.mutable.Map[ClientTextId, CompleteClientText]()
|
||||
while rs.next() do {
|
||||
val id = rs.getString("id")
|
||||
result(id) = CompleteClientText(
|
||||
id = id,
|
||||
text = rs.getString("text"),
|
||||
requestedAfterHistoryCount = rs.getInt("requested_after_history_count")
|
||||
)
|
||||
}
|
||||
event.rows = result.size
|
||||
result.toMap
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
override def incompleteTexts: Map[ClientTextId, IncompleteClientText] = {
|
||||
val sql =
|
||||
"SELECT id, text, llm_request, requested_after_history_count, requested_at_millis, last_update_at_millis FROM client_texts WHERE status = 'incomplete'"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.incompleteTexts", sql) { event =>
|
||||
try {
|
||||
val rs = stmt.executeQuery()
|
||||
val result = scala.collection.mutable.Map[ClientTextId, IncompleteClientText]()
|
||||
while rs.next() do {
|
||||
val id = rs.getString("id")
|
||||
val llmRequest = llmRequestFromResultSet(rs)
|
||||
result(id) = IncompleteClientText(
|
||||
id = id,
|
||||
partialText = rs.getString("text"),
|
||||
llmRequest = llmRequest,
|
||||
requestedAfterHistoryCount = rs.getInt("requested_after_history_count"),
|
||||
requestedAtMillis = rs.getLong("requested_at_millis"),
|
||||
lastUpdateAtMillis = rs.getLong("last_update_at_millis")
|
||||
)
|
||||
}
|
||||
event.rows = result.size
|
||||
result.toMap
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
override def unrequestedTexts: Map[ClientTextId, UnrequestedClientText] = {
|
||||
val sql = "SELECT id, llm_request, requested_after_history_count FROM client_texts WHERE status = 'unrequested'"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.unrequestedTexts", sql) { event =>
|
||||
try {
|
||||
val rs = stmt.executeQuery()
|
||||
val result = scala.collection.mutable.Map[ClientTextId, UnrequestedClientText]()
|
||||
while rs.next() do {
|
||||
val id = rs.getString("id")
|
||||
result(id) = UnrequestedClientText(
|
||||
id = id,
|
||||
llmRequest = llmRequestFromResultSet(rs),
|
||||
requestedAfterHistoryCount = rs.getInt("requested_after_history_count")
|
||||
)
|
||||
}
|
||||
event.rows = result.size
|
||||
result.toMap
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
override def accessibleTo: Map[ClientTextId, Vector[FactionId]] = {
|
||||
val sql = "SELECT text_id, faction_id FROM client_text_visibility ORDER BY text_id, faction_id"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.accessibleTo", sql) { event =>
|
||||
try {
|
||||
val rs = stmt.executeQuery()
|
||||
val result = scala.collection.mutable.Map[ClientTextId, Vector[FactionId]]()
|
||||
var rows = 0
|
||||
while rs.next() do {
|
||||
val textId = rs.getString("text_id")
|
||||
val factionId = rs.getInt("faction_id")
|
||||
result(textId) = result.getOrElse(textId, Vector.empty) :+ factionId
|
||||
rows += 1
|
||||
}
|
||||
event.rows = rows
|
||||
result.toMap
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
override def accessibleToForId(id: ClientTextId): Vector[FactionId] = {
|
||||
val sql = "SELECT faction_id FROM client_text_visibility WHERE text_id = ? ORDER BY faction_id"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.accessibleToForId", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, id)
|
||||
val rs = stmt.executeQuery()
|
||||
val result = scala.collection.mutable.ArrayBuffer[FactionId]()
|
||||
while rs.next() do result += rs.getInt("faction_id")
|
||||
event.rows = result.size
|
||||
result.toVector
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
override def beginTransaction(): Unit = {
|
||||
if !connection.getAutoCommit then return
|
||||
connection.setAutoCommit(false)
|
||||
}
|
||||
|
||||
override def commitTransaction(): Unit = {
|
||||
if connection.getAutoCommit then return
|
||||
connection.commit()
|
||||
connection.setAutoCommit(true)
|
||||
}
|
||||
|
||||
override def rollbackTransaction(): Unit = {
|
||||
if connection.getAutoCommit then return
|
||||
connection.rollback()
|
||||
connection.setAutoCommit(true)
|
||||
}
|
||||
|
||||
override def saved: ClientTextStore = this
|
||||
|
||||
override def withAddedTextRequest(
|
||||
id: ClientTextId,
|
||||
accessibleTo: Vector[FactionId],
|
||||
llmRequest: GeneratedTextRequestT,
|
||||
requestedAfterHistoryCount: Int
|
||||
): ClientTextStore = {
|
||||
internalRequire(
|
||||
pregenerated.getText(id).isEmpty,
|
||||
s"Text with id $id is pregenerated"
|
||||
)
|
||||
|
||||
if textExists(id) then {
|
||||
println(s"Text with id $id already exists")
|
||||
return this
|
||||
}
|
||||
|
||||
val sql =
|
||||
"""INSERT INTO client_texts (id, status, text, llm_request, requested_after_history_count)
|
||||
|VALUES (?, 'unrequested', '', ?, ?)""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.withAddedTextRequest.insertText", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, id)
|
||||
stmt.setBytes(2, GeneratedTextRequestConverter.toProto(llmRequest).toByteArray)
|
||||
stmt.setInt(3, requestedAfterHistoryCount)
|
||||
event.rows = stmt.executeUpdate()
|
||||
} finally stmt.close()
|
||||
}
|
||||
|
||||
insertVisibility(id, accessibleTo)
|
||||
this
|
||||
}
|
||||
|
||||
override def getText(id: ClientTextId): TextGenerationResult =
|
||||
pregenerated
|
||||
.getText(id)
|
||||
.map(text => TextGenerationSuccess(text))
|
||||
.getOrElse {
|
||||
val sql = "SELECT status, text FROM client_texts WHERE id = ?"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.getText", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, id)
|
||||
val rs = stmt.executeQuery()
|
||||
if rs.next() then {
|
||||
event.rows = 1
|
||||
val status = rs.getString("status")
|
||||
val text = rs.getString("text")
|
||||
status match {
|
||||
case "complete" => TextGenerationSuccess(text)
|
||||
case "incomplete" => TextGenerationDependencyInProgress(id, text)
|
||||
case "unrequested" => TextGenerationDependencyWaiting(id)
|
||||
}
|
||||
} else TextGenerationDependencyUnknown(id)
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
override def withMarkedRequested(id: ClientTextId): ClientTextStore = {
|
||||
val now = System.currentTimeMillis()
|
||||
val sql =
|
||||
"""UPDATE client_texts SET status = 'incomplete', requested_at_millis = ?, last_update_at_millis = ?
|
||||
|WHERE id = ? AND status = 'unrequested'""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.withMarkedRequested", sql) { event =>
|
||||
try {
|
||||
stmt.setLong(1, now)
|
||||
stmt.setLong(2, now)
|
||||
stmt.setString(3, id)
|
||||
val updated = stmt.executeUpdate()
|
||||
event.rows = updated
|
||||
if updated == 0 then {
|
||||
throw new EagleInternalException(
|
||||
s"Unrequested text with id $id not found"
|
||||
)
|
||||
}
|
||||
} finally stmt.close()
|
||||
}
|
||||
this
|
||||
}
|
||||
|
||||
override def withBypassed(id: ClientTextId): ClientTextStore = {
|
||||
val sql = "DELETE FROM client_texts WHERE id = ? AND status = 'unrequested'"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.withBypassed", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, id)
|
||||
event.rows = stmt.executeUpdate()
|
||||
} finally stmt.close()
|
||||
}
|
||||
this
|
||||
}
|
||||
|
||||
override def withAppendedText(
|
||||
id: ClientTextId,
|
||||
newText: String,
|
||||
complete: Boolean
|
||||
): ClientTextStoreWithUpdate = {
|
||||
internalRequire(
|
||||
pregenerated.getText(id).isEmpty,
|
||||
s"Text with id $id is pregenerated"
|
||||
)
|
||||
|
||||
val (currentText, llmRequest, requestedAfterHistoryCount, requestedAtMillis) =
|
||||
selectIncompleteText(id)
|
||||
val now = System.currentTimeMillis()
|
||||
val updatedText = currentText + newText
|
||||
val newStatus = if complete then "complete" else "incomplete"
|
||||
|
||||
val sql =
|
||||
"""UPDATE client_texts SET text = ?, status = ?, last_update_at_millis = ?
|
||||
|WHERE id = ?""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.withAppendedText", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, updatedText)
|
||||
stmt.setString(2, newStatus)
|
||||
stmt.setLong(3, now)
|
||||
stmt.setString(4, id)
|
||||
event.rows = stmt.executeUpdate()
|
||||
} finally stmt.close()
|
||||
}
|
||||
|
||||
val updatedClientText =
|
||||
if complete then
|
||||
CompleteClientText(
|
||||
id = id,
|
||||
text = updatedText,
|
||||
requestedAfterHistoryCount = requestedAfterHistoryCount
|
||||
)
|
||||
else
|
||||
IncompleteClientText(
|
||||
id = id,
|
||||
partialText = updatedText,
|
||||
llmRequest = llmRequest,
|
||||
requestedAfterHistoryCount = requestedAfterHistoryCount,
|
||||
requestedAtMillis = requestedAtMillis,
|
||||
lastUpdateAtMillis = now
|
||||
)
|
||||
|
||||
ClientTextStoreWithUpdate(this, updatedClientText)
|
||||
}
|
||||
|
||||
override def withExtendedVisibility(
|
||||
id: ClientTextId,
|
||||
addedFactionIds: Vector[FactionId]
|
||||
): ClientTextStore = {
|
||||
insertVisibility(id, addedFactionIds)
|
||||
this
|
||||
}
|
||||
|
||||
override def withMovedBackToUnrequested(id: ClientTextId): ClientTextStore = {
|
||||
val status = statusFor(id)
|
||||
if status.isEmpty then {
|
||||
println(s"Warning: Attempted to move non-existent text $id back to unrequested")
|
||||
return this
|
||||
}
|
||||
if !status.contains("incomplete") then {
|
||||
println(s"Warning: Attempted to move non-incomplete text $id back to unrequested")
|
||||
return this
|
||||
}
|
||||
|
||||
val sql =
|
||||
"""UPDATE client_texts SET status = 'unrequested', text = '', requested_at_millis = NULL, last_update_at_millis = NULL
|
||||
|WHERE id = ? AND status = 'incomplete'""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.withMovedBackToUnrequested", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, id)
|
||||
event.rows = stmt.executeUpdate()
|
||||
} finally stmt.close()
|
||||
}
|
||||
this
|
||||
}
|
||||
|
||||
override def withAddedCompleteText(
|
||||
id: ClientTextId,
|
||||
text: String,
|
||||
accessibleTo: Vector[FactionId]
|
||||
): ClientTextStore = {
|
||||
if textExists(id) then return this
|
||||
|
||||
val sql =
|
||||
"""INSERT INTO client_texts (id, status, text, requested_after_history_count)
|
||||
|VALUES (?, 'complete', ?, 0)""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.withAddedCompleteText", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, id)
|
||||
stmt.setString(2, text)
|
||||
event.rows = stmt.executeUpdate()
|
||||
} finally stmt.close()
|
||||
}
|
||||
insertVisibility(id, accessibleTo)
|
||||
this
|
||||
}
|
||||
|
||||
override def withAddedCompleteTexts(
|
||||
texts: Vector[(ClientTextId, String)],
|
||||
accessibleTo: Vector[FactionId]
|
||||
): ClientTextStore = {
|
||||
if texts.isEmpty then return this
|
||||
|
||||
val originalAutoCommit = connection.getAutoCommit
|
||||
connection.setAutoCommit(false)
|
||||
try {
|
||||
val existingIds = readExistingIds()
|
||||
val newTexts = texts.filterNot { case (id, _) => existingIds.contains(id) }
|
||||
if newTexts.isEmpty then {
|
||||
connection.setAutoCommit(originalAutoCommit)
|
||||
return this
|
||||
}
|
||||
|
||||
val insertSql =
|
||||
"""INSERT INTO client_texts (id, status, text, requested_after_history_count)
|
||||
|VALUES (?, 'complete', ?, 0)""".stripMargin
|
||||
val insertStmt = connection.prepareStatement(insertSql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.withAddedCompleteTexts.insertTexts", insertSql) {
|
||||
event =>
|
||||
try {
|
||||
newTexts.foreach {
|
||||
case (id, text) =>
|
||||
insertStmt.setString(1, id)
|
||||
insertStmt.setString(2, text)
|
||||
insertStmt.addBatch()
|
||||
}
|
||||
event.rows = insertStmt.executeBatch().sum
|
||||
} finally insertStmt.close()
|
||||
}
|
||||
|
||||
insertVisibilityBatch(newTexts.map(_._1), accessibleTo)
|
||||
connection.commit()
|
||||
this
|
||||
} catch {
|
||||
case e: Exception =>
|
||||
connection.rollback()
|
||||
throw e
|
||||
} finally connection.setAutoCommit(originalAutoCommit)
|
||||
end try
|
||||
}
|
||||
|
||||
override def withCachedUnrequestedGameState(
|
||||
id: ClientTextId,
|
||||
updatedEntry: UnrequestedClientText
|
||||
): ClientTextStore = this
|
||||
|
||||
override def withCachedIncompleteGameState(
|
||||
id: ClientTextId,
|
||||
updatedEntry: IncompleteClientText
|
||||
): ClientTextStore = this
|
||||
|
||||
override def hasIncompleteTextsAccessibleTo(factionId: FactionId): Boolean = {
|
||||
val sql =
|
||||
"""SELECT EXISTS(
|
||||
| SELECT 1 FROM client_texts t
|
||||
| LEFT JOIN client_text_visibility v ON t.id = v.text_id
|
||||
| WHERE t.status IN ('incomplete', 'unrequested')
|
||||
| AND (v.faction_id IS NULL OR v.faction_id = ?)
|
||||
|)""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.hasIncompleteTextsAccessibleTo", sql) { event =>
|
||||
try {
|
||||
stmt.setInt(1, factionId)
|
||||
val rs = stmt.executeQuery()
|
||||
val result = rs.next() && rs.getBoolean(1)
|
||||
event.rows = if result then 1 else 0
|
||||
result
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
override def getCompleteTextsAccessibleTo(factionId: FactionId): Vector[CompleteClientText] = {
|
||||
val sql =
|
||||
"""SELECT t.id, t.text, t.requested_after_history_count
|
||||
|FROM client_texts t
|
||||
|WHERE t.status = 'complete'
|
||||
|AND (
|
||||
| NOT EXISTS (SELECT 1 FROM client_text_visibility v WHERE v.text_id = t.id)
|
||||
| OR EXISTS (SELECT 1 FROM client_text_visibility v WHERE v.text_id = t.id AND v.faction_id = ?)
|
||||
|)""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.getCompleteTextsAccessibleTo", sql) { event =>
|
||||
try {
|
||||
stmt.setInt(1, factionId)
|
||||
val rs = stmt.executeQuery()
|
||||
val result = scala.collection.mutable.ArrayBuffer[CompleteClientText]()
|
||||
while rs.next() do
|
||||
result += CompleteClientText(
|
||||
id = rs.getString("id"),
|
||||
text = rs.getString("text"),
|
||||
requestedAfterHistoryCount = rs.getInt("requested_after_history_count")
|
||||
)
|
||||
event.rows = result.size
|
||||
result.toVector
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
def hasAnyTexts: Boolean = {
|
||||
val sql = "SELECT EXISTS(SELECT 1 FROM client_texts)"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.hasAnyTexts", sql) { event =>
|
||||
try {
|
||||
val rs = stmt.executeQuery()
|
||||
val result = rs.next() && rs.getBoolean(1)
|
||||
event.rows = if result then 1 else 0
|
||||
result
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
def importFrom(source: ClientTextStore): Unit = {
|
||||
val originalAutoCommit = connection.getAutoCommit
|
||||
connection.setAutoCommit(false)
|
||||
try {
|
||||
importCompleteTexts(source.completeTexts.values.toVector)
|
||||
importIncompleteTexts(source.incompleteTexts.values.toVector)
|
||||
importUnrequestedTexts(source.unrequestedTexts.values.toVector)
|
||||
source.accessibleTo.foreach {
|
||||
case (textId, factionIds) =>
|
||||
insertVisibility(textId, factionIds)
|
||||
}
|
||||
connection.commit()
|
||||
} catch {
|
||||
case e: Exception =>
|
||||
connection.rollback()
|
||||
throw e
|
||||
} finally connection.setAutoCommit(originalAutoCommit)
|
||||
}
|
||||
|
||||
private def importCompleteTexts(texts: Vector[CompleteClientText]): Unit = {
|
||||
if texts.isEmpty then return
|
||||
|
||||
val sql =
|
||||
"""INSERT INTO client_texts (id, status, text, requested_after_history_count)
|
||||
|VALUES (?, 'complete', ?, ?)
|
||||
|ON CONFLICT DO NOTHING""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.importCompleteTexts", sql) { event =>
|
||||
try {
|
||||
texts.foreach { text =>
|
||||
stmt.setString(1, text.id)
|
||||
stmt.setString(2, text.text)
|
||||
stmt.setInt(3, text.requestedAfterHistoryCount)
|
||||
stmt.addBatch()
|
||||
}
|
||||
event.rows = stmt.executeBatch().sum
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
private def importIncompleteTexts(texts: Vector[IncompleteClientText]): Unit = {
|
||||
if texts.isEmpty then return
|
||||
|
||||
val sql =
|
||||
"""INSERT INTO client_texts
|
||||
| (id, status, text, llm_request, requested_after_history_count, requested_at_millis, last_update_at_millis)
|
||||
|VALUES (?, 'incomplete', ?, ?, ?, ?, ?)
|
||||
|ON CONFLICT DO NOTHING""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.importIncompleteTexts", sql) { event =>
|
||||
try {
|
||||
texts.foreach { text =>
|
||||
stmt.setString(1, text.id)
|
||||
stmt.setString(2, text.partialText)
|
||||
stmt.setBytes(3, GeneratedTextRequestConverter.toProto(text.llmRequest).toByteArray)
|
||||
stmt.setInt(4, text.requestedAfterHistoryCount)
|
||||
stmt.setLong(5, text.requestedAtMillis)
|
||||
stmt.setLong(6, text.lastUpdateAtMillis)
|
||||
stmt.addBatch()
|
||||
}
|
||||
event.rows = stmt.executeBatch().sum
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
private def importUnrequestedTexts(texts: Vector[UnrequestedClientText]): Unit = {
|
||||
if texts.isEmpty then return
|
||||
|
||||
val sql =
|
||||
"""INSERT INTO client_texts (id, status, text, llm_request, requested_after_history_count)
|
||||
|VALUES (?, 'unrequested', '', ?, ?)
|
||||
|ON CONFLICT DO NOTHING""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.importUnrequestedTexts", sql) { event =>
|
||||
try {
|
||||
texts.foreach { text =>
|
||||
stmt.setString(1, text.id)
|
||||
stmt.setBytes(2, GeneratedTextRequestConverter.toProto(text.llmRequest).toByteArray)
|
||||
stmt.setInt(3, text.requestedAfterHistoryCount)
|
||||
stmt.addBatch()
|
||||
}
|
||||
event.rows = stmt.executeBatch().sum
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
private def llmRequestFromResultSet(rs: ResultSet): GeneratedTextRequestT =
|
||||
GeneratedTextRequestConverter.fromProto(
|
||||
net.eagle0.eagle.internal.generated_text_request.GeneratedTextRequest.parseFrom(rs.getBytes("llm_request"))
|
||||
)
|
||||
|
||||
private def textExists(id: ClientTextId): Boolean =
|
||||
statusFor(id).nonEmpty
|
||||
|
||||
private def statusFor(id: ClientTextId): Option[String] = {
|
||||
val sql = "SELECT status FROM client_texts WHERE id = ?"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.statusFor", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, id)
|
||||
val rs = stmt.executeQuery()
|
||||
if rs.next() then {
|
||||
event.rows = 1
|
||||
Some(rs.getString("status"))
|
||||
} else None
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
private def selectIncompleteText(
|
||||
id: ClientTextId
|
||||
): (String, GeneratedTextRequestT, Int, Long) = {
|
||||
val sql =
|
||||
"SELECT status, text, llm_request, requested_after_history_count, requested_at_millis FROM client_texts WHERE id = ?"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.selectIncompleteText", sql) { event =>
|
||||
try {
|
||||
stmt.setString(1, id)
|
||||
val rs = stmt.executeQuery()
|
||||
if !rs.next() then throw new EagleInternalException(s"Text with id $id not found")
|
||||
event.rows = 1
|
||||
val status = rs.getString("status")
|
||||
internalRequire(
|
||||
status == "incomplete",
|
||||
s"Cannot append to text with status $status (expected incomplete)"
|
||||
)
|
||||
(
|
||||
rs.getString("text"),
|
||||
llmRequestFromResultSet(rs),
|
||||
rs.getInt("requested_after_history_count"),
|
||||
rs.getLong("requested_at_millis")
|
||||
)
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
private def readExistingIds(): Set[ClientTextId] = {
|
||||
val sql = "SELECT id FROM client_texts"
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.readExistingIds", sql) { event =>
|
||||
try {
|
||||
val rs = stmt.executeQuery()
|
||||
val result = scala.collection.mutable.Set[ClientTextId]()
|
||||
while rs.next() do result += rs.getString("id")
|
||||
event.rows = result.size
|
||||
result.toSet
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
|
||||
private def insertVisibility(id: ClientTextId, factionIds: Vector[FactionId]): Unit =
|
||||
insertVisibilityBatch(Vector(id), factionIds)
|
||||
|
||||
private def insertVisibilityBatch(ids: Vector[ClientTextId], factionIds: Vector[FactionId]): Unit = {
|
||||
if ids.isEmpty || factionIds.isEmpty then return
|
||||
|
||||
val sql =
|
||||
"""INSERT INTO client_text_visibility (text_id, faction_id)
|
||||
|VALUES (?, ?)
|
||||
|ON CONFLICT DO NOTHING""".stripMargin
|
||||
val stmt = connection.prepareStatement(sql)
|
||||
JfrEvents.postgresQuery(databaseName, "PostgresClientTextStore.insertVisibilityBatch", sql) { event =>
|
||||
try {
|
||||
for
|
||||
id <- ids
|
||||
factionId <- factionIds
|
||||
do {
|
||||
stmt.setString(1, id)
|
||||
stmt.setInt(2, factionId)
|
||||
stmt.addBatch()
|
||||
}
|
||||
event.rows = stmt.executeBatch().sum
|
||||
} finally stmt.close()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
object PostgresClientTextStore {
|
||||
private val SchemaVersion = 1
|
||||
|
||||
def createWithData(
|
||||
gameId: GameId,
|
||||
pregenerated: PregeneratedClientTextStore,
|
||||
unrequestedTexts: Map[ClientTextId, UnrequestedClientText],
|
||||
accessibleTo: Map[ClientTextId, Vector[FactionId]]
|
||||
): PostgresClientTextStore = {
|
||||
val store = create(gameId, pregenerated)
|
||||
val originalAutoCommit = store.connection.getAutoCommit
|
||||
store.connection.setAutoCommit(false)
|
||||
try {
|
||||
val insertSql =
|
||||
"""INSERT INTO client_texts (id, status, text, llm_request, requested_after_history_count)
|
||||
|VALUES (?, 'unrequested', '', ?, ?)""".stripMargin
|
||||
val insertStmt = store.connection.prepareStatement(insertSql)
|
||||
JfrEvents.postgresQuery(store.databaseName, "PostgresClientTextStore.createWithData.insertTexts", insertSql) {
|
||||
event =>
|
||||
try {
|
||||
unrequestedTexts.values.foreach { ut =>
|
||||
insertStmt.setString(1, ut.id)
|
||||
insertStmt.setBytes(2, GeneratedTextRequestConverter.toProto(ut.llmRequest).toByteArray)
|
||||
insertStmt.setInt(3, ut.requestedAfterHistoryCount)
|
||||
insertStmt.addBatch()
|
||||
}
|
||||
event.rows = insertStmt.executeBatch().sum
|
||||
} finally insertStmt.close()
|
||||
}
|
||||
|
||||
accessibleTo.foreach {
|
||||
case (textId, factionIds) =>
|
||||
store.insertVisibility(textId, factionIds)
|
||||
}
|
||||
store.connection.commit()
|
||||
store
|
||||
} catch {
|
||||
case e: Exception =>
|
||||
store.connection.rollback()
|
||||
throw e
|
||||
} finally store.connection.setAutoCommit(originalAutoCommit)
|
||||
end try
|
||||
}
|
||||
|
||||
def loaded(
|
||||
gameId: GameId,
|
||||
pregenerated: PregeneratedClientTextStore
|
||||
): Try[PostgresClientTextStore] = Try {
|
||||
val connection = openConnection()
|
||||
try {
|
||||
val schemaName = schemaNameForGame(gameId)
|
||||
createSchema(connection, schemaName)
|
||||
useSchema(connection, schemaName)
|
||||
initializeSchema(connection)
|
||||
new PostgresClientTextStore(connection, gameId, pregenerated)
|
||||
} catch {
|
||||
case t: Throwable =>
|
||||
connection.close()
|
||||
throw t
|
||||
}
|
||||
}
|
||||
|
||||
private def create(
|
||||
gameId: GameId,
|
||||
pregenerated: PregeneratedClientTextStore
|
||||
): PostgresClientTextStore =
|
||||
loaded(gameId, pregenerated).get
|
||||
|
||||
private def openConnection(): Connection = {
|
||||
val cfg = config.getOrElse(
|
||||
throw new EagleInternalException(
|
||||
"PostgresClientTextStore requires EAGLE_POSTGRES_HOST, DATABASE, USER, and PASSWORD"
|
||||
)
|
||||
)
|
||||
Class.forName("org.postgresql.Driver")
|
||||
DriverManager.getConnection(cfg.jdbcUrl, cfg.user, cfg.password)
|
||||
}
|
||||
|
||||
private def config: Option[PostgresConfig] =
|
||||
for
|
||||
host <- envNonEmpty("EAGLE_POSTGRES_HOST")
|
||||
database <- envNonEmpty("EAGLE_POSTGRES_DATABASE")
|
||||
user <- envNonEmpty("EAGLE_POSTGRES_USER")
|
||||
password <- envNonEmpty("EAGLE_POSTGRES_PASSWORD")
|
||||
yield PostgresConfig(
|
||||
host = host,
|
||||
port = sys.env.get("EAGLE_POSTGRES_PORT").flatMap(_.toIntOption).getOrElse(5432),
|
||||
database = database,
|
||||
user = user,
|
||||
password = password,
|
||||
sslmode = sys.env.get("EAGLE_POSTGRES_SSLMODE").filter(_.nonEmpty).getOrElse("require")
|
||||
)
|
||||
|
||||
private def envNonEmpty(name: String): Option[String] =
|
||||
sys.env.get(name).map(_.trim).filter(_.nonEmpty)
|
||||
|
||||
private def schemaNameForGame(gameId: GameId): String = {
|
||||
val schema = "eagle_game_" + java.lang.Long.toHexString(gameId)
|
||||
if !schema.matches("[a-z][a-z0-9_]*") then
|
||||
throw new EagleInternalException(s"Invalid Postgres schema name derived from game id $gameId: $schema")
|
||||
schema
|
||||
}
|
||||
|
||||
private def createSchema(connection: Connection, schemaName: String): Unit = {
|
||||
val stmt = connection.createStatement()
|
||||
try stmt.execute(s"CREATE SCHEMA IF NOT EXISTS $schemaName"): Unit
|
||||
finally stmt.close()
|
||||
}
|
||||
|
||||
private def useSchema(connection: Connection, schemaName: String): Unit = {
|
||||
val stmt = connection.createStatement()
|
||||
try stmt.execute(s"SET search_path TO $schemaName"): Unit
|
||||
finally stmt.close()
|
||||
}
|
||||
|
||||
private def initializeSchema(connection: Connection): Unit = {
|
||||
execEach(
|
||||
connection,
|
||||
Vector("CREATE TABLE IF NOT EXISTS client_text_metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
|
||||
)
|
||||
val version = readSchemaVersion(connection)
|
||||
if version == SchemaVersion then ()
|
||||
else if version == 0 then {
|
||||
val originalAutoCommit = connection.getAutoCommit
|
||||
connection.setAutoCommit(false)
|
||||
try {
|
||||
execEach(connection, schemaStatements)
|
||||
writeSchemaVersion(connection, SchemaVersion)
|
||||
connection.commit()
|
||||
} catch {
|
||||
case t: Throwable =>
|
||||
connection.rollback()
|
||||
throw t
|
||||
} finally connection.setAutoCommit(originalAutoCommit)
|
||||
} else
|
||||
throw new EagleInternalException(
|
||||
s"Unsupported Postgres client text schema version $version; expected $SchemaVersion"
|
||||
)
|
||||
end if
|
||||
}
|
||||
|
||||
private def readSchemaVersion(connection: Connection): Int = {
|
||||
val stmt = connection.prepareStatement("SELECT value FROM client_text_metadata WHERE key = 'schema_version'")
|
||||
try {
|
||||
val rs = stmt.executeQuery()
|
||||
if rs.next() then rs.getString(1).toInt else 0
|
||||
} finally stmt.close()
|
||||
}
|
||||
|
||||
private def writeSchemaVersion(connection: Connection, version: Int): Unit = {
|
||||
val stmt = connection.prepareStatement(
|
||||
"""INSERT INTO client_text_metadata (key, value) VALUES ('schema_version', ?)
|
||||
|ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value""".stripMargin
|
||||
)
|
||||
try {
|
||||
stmt.setString(1, version.toString)
|
||||
stmt.executeUpdate(): Unit
|
||||
} finally stmt.close()
|
||||
}
|
||||
|
||||
private def execEach(connection: Connection, statements: Vector[String]): Unit = {
|
||||
val stmt = connection.createStatement()
|
||||
try statements.foreach(sql => stmt.execute(sql))
|
||||
finally stmt.close()
|
||||
}
|
||||
|
||||
private def schemaStatements: Vector[String] =
|
||||
Vector(
|
||||
"""CREATE TABLE client_texts (
|
||||
| id TEXT PRIMARY KEY,
|
||||
| status TEXT NOT NULL CHECK (status IN ('complete', 'incomplete', 'unrequested')),
|
||||
| text TEXT NOT NULL DEFAULT '',
|
||||
| llm_request BYTEA,
|
||||
| requested_after_history_count INTEGER NOT NULL,
|
||||
| requested_at_millis BIGINT,
|
||||
| last_update_at_millis BIGINT
|
||||
|)""".stripMargin,
|
||||
"""CREATE TABLE client_text_visibility (
|
||||
| text_id TEXT NOT NULL,
|
||||
| faction_id INTEGER NOT NULL,
|
||||
| PRIMARY KEY (text_id, faction_id)
|
||||
|)""".stripMargin,
|
||||
"CREATE INDEX idx_client_texts_status ON client_texts(status)"
|
||||
)
|
||||
|
||||
private case class PostgresConfig(
|
||||
host: String,
|
||||
port: Int,
|
||||
database: String,
|
||||
user: String,
|
||||
password: String,
|
||||
sslmode: String
|
||||
) {
|
||||
def jdbcUrl: String =
|
||||
s"jdbc:postgresql://$host:$port/$database?sslmode=$sslmode"
|
||||
}
|
||||
}
|
||||
@@ -277,6 +277,7 @@ scala_library(
|
||||
"//src/main/scala/net/eagle0/eagle/client_text",
|
||||
"//src/main/scala/net/eagle0/eagle/client_text:client_text_store",
|
||||
"//src/main/scala/net/eagle0/eagle/client_text:client_text_store_impl",
|
||||
"//src/main/scala/net/eagle0/eagle/client_text:postgres_client_text_store",
|
||||
"//src/main/scala/net/eagle0/eagle/client_text:sqlite_client_text_store",
|
||||
"//src/main/scala/net/eagle0/eagle/client_text:text_generation_result",
|
||||
"//src/main/scala/net/eagle0/eagle/library:eagle_internal_exception",
|
||||
|
||||
@@ -9,7 +9,7 @@ import io.sentry.Sentry
|
||||
import net.eagle0.common.{ProtoParser, SeededRandom, SimpleTimedLogger, ZipUtils}
|
||||
import net.eagle0.common.shardok_internal_interface.ShardokInternalInterfaceGrpc.ShardokInternalInterfaceStub
|
||||
import net.eagle0.common.tutorial_battle_config.TutorialBattleConfig
|
||||
import net.eagle0.eagle.{FactionId, GameId, ProvinceId, ShardokGameId, UserId}
|
||||
import net.eagle0.eagle.{ClientTextId, FactionId, GameId, ProvinceId, ShardokGameId, UserId}
|
||||
import net.eagle0.eagle.admin.game_admin.{ConvertAiToHumanResponse, ConvertHumanToAiResponse, ReassignFactionResponse}
|
||||
import net.eagle0.eagle.api.eagle.*
|
||||
import net.eagle0.eagle.api.eagle.JoinGameResult.INVALID_OPTIONS_RESULT
|
||||
@@ -19,6 +19,7 @@ import net.eagle0.eagle.api.selected_command.SelectedCommand
|
||||
import net.eagle0.eagle.auth.UserService
|
||||
import net.eagle0.eagle.client_text.{
|
||||
ClientTextStore,
|
||||
PostgresClientTextStore,
|
||||
PregeneratedClientTextStore,
|
||||
SqliteClientTextStore,
|
||||
TextGenerationSuccess,
|
||||
@@ -224,6 +225,80 @@ object GamesManager {
|
||||
)
|
||||
}
|
||||
|
||||
private def loadedClientTextStore(
|
||||
gameId: GameId,
|
||||
databasePersister: Option[Persister],
|
||||
pregenerated: PregeneratedClientTextStore
|
||||
): ClientTextStore =
|
||||
if HistoryBackendConfig.usePostgres then {
|
||||
val postgresStore = PostgresClientTextStore
|
||||
.loaded(
|
||||
gameId = gameId,
|
||||
pregenerated = pregenerated
|
||||
)
|
||||
.get
|
||||
importLocalSqliteTextStoreIfNeeded(
|
||||
gameId = gameId,
|
||||
postgresStore = postgresStore,
|
||||
databasePersister = databasePersister,
|
||||
pregenerated = pregenerated
|
||||
)
|
||||
postgresStore
|
||||
} else
|
||||
SqliteClientTextStore
|
||||
.loaded(
|
||||
saveDirectory = new java.io.File(SaveDirectory.saveDirectoryForGame(gameId)),
|
||||
pregenerated = pregenerated,
|
||||
persister = databasePersister
|
||||
)
|
||||
.get
|
||||
|
||||
private def importLocalSqliteTextStoreIfNeeded(
|
||||
gameId: GameId,
|
||||
postgresStore: PostgresClientTextStore,
|
||||
databasePersister: Option[Persister],
|
||||
pregenerated: PregeneratedClientTextStore
|
||||
): Unit = {
|
||||
val saveDirectory = new java.io.File(SaveDirectory.saveDirectoryForGame(gameId))
|
||||
val sqliteDbFile = new java.io.File(saveDirectory, "text_store.db")
|
||||
if !postgresStore.hasAnyTexts && sqliteDbFile.exists() then {
|
||||
val sqliteStore = SqliteClientTextStore
|
||||
.loaded(
|
||||
saveDirectory = saveDirectory,
|
||||
pregenerated = pregenerated,
|
||||
persister = databasePersister
|
||||
)
|
||||
.get
|
||||
postgresStore.importFrom(sqliteStore)
|
||||
println(
|
||||
s"Imported local text_store.db into Postgres text store for game ${gameId.toHexString}"
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
private def createClientTextStoreWithData(
|
||||
gameId: GameId,
|
||||
pregenerated: PregeneratedClientTextStore,
|
||||
unrequestedTexts: Map[ClientTextId, UnrequestedClientText],
|
||||
accessibleTo: Map[ClientTextId, Vector[FactionId]],
|
||||
databasePersister: Option[Persister]
|
||||
): ClientTextStore =
|
||||
if HistoryBackendConfig.usePostgres then
|
||||
PostgresClientTextStore.createWithData(
|
||||
gameId = gameId,
|
||||
pregenerated = pregenerated,
|
||||
unrequestedTexts = unrequestedTexts,
|
||||
accessibleTo = accessibleTo
|
||||
)
|
||||
else
|
||||
SqliteClientTextStore.createWithData(
|
||||
dbPath = new java.io.File(SaveDirectory.saveDirectoryForGame(gameId), "text_store.db"),
|
||||
pregenerated = pregenerated,
|
||||
unrequestedTexts = unrequestedTexts,
|
||||
accessibleTo = accessibleTo,
|
||||
persister = databasePersister
|
||||
)
|
||||
|
||||
private def gameController(
|
||||
gameId: GameId,
|
||||
databasePersister: Option[Persister],
|
||||
@@ -232,13 +307,11 @@ object GamesManager {
|
||||
pregeneratedHeroes: Vector[HeroWithName]
|
||||
): GameController = {
|
||||
val textStoreStart = System.currentTimeMillis()
|
||||
val loadedTextStore = SqliteClientTextStore
|
||||
.loaded(
|
||||
saveDirectory = new java.io.File(SaveDirectory.saveDirectoryForGame(gameId)),
|
||||
pregenerated = pregeneratedTexts,
|
||||
persister = databasePersister
|
||||
)
|
||||
.get
|
||||
val loadedTextStore = GamesManager.loadedClientTextStore(
|
||||
gameId = gameId,
|
||||
databasePersister = databasePersister,
|
||||
pregenerated = pregeneratedTexts
|
||||
)
|
||||
val textStoreDuration = System.currentTimeMillis() - textStoreStart
|
||||
|
||||
// Check all heroes' nameTextIds and recover any missing name requests
|
||||
@@ -1540,16 +1613,12 @@ class GamesManager(
|
||||
if startPhase == TutorialPhase.OpeningBattle then
|
||||
tutorialBattleConfigs = tutorialBattleConfigs + (gameId -> tutorialConfig)
|
||||
|
||||
// Use the SQLite-backed store, matching normal game creation. The reload path
|
||||
// (gameController -> SqliteClientTextStore.loaded) always reads text_store.db, so the
|
||||
// tutorial must persist there too — otherwise texts generated mid-tutorial (e.g. yearly
|
||||
// chronicle updates) are lost on the first save+reload.
|
||||
val initialTextStore = SqliteClientTextStore.createWithData(
|
||||
dbPath = new java.io.File(SaveDirectory.saveDirectoryForGame(gameId), "text_store.db"),
|
||||
val initialTextStore = GamesManager.createClientTextStoreWithData(
|
||||
gameId = gameId,
|
||||
pregenerated = pregeneratedClientText,
|
||||
unrequestedTexts = Map.empty,
|
||||
accessibleTo = Map.empty,
|
||||
persister = gamePersisterCreation.databaseUploadPersisterForGame(gameId)
|
||||
databasePersister = gamePersisterCreation.databaseUploadPersisterForGame(gameId)
|
||||
)
|
||||
|
||||
// Copy pregenerated texts (hero names, backstories) into the game's text store
|
||||
@@ -1684,16 +1753,16 @@ class GamesManager(
|
||||
.toMap
|
||||
|
||||
// Create the text store and populate it with pregenerated texts for heroes in this game
|
||||
val initialTextStore = logger.withStopwatch(s"[CREATE:$gameIdHex] SqliteClientTextStore.createWithData") {
|
||||
SqliteClientTextStore.createWithData(
|
||||
dbPath = new java.io.File(SaveDirectory.saveDirectoryForGame(gameId), "text_store.db"),
|
||||
val initialTextStore = logger.withStopwatch(s"[CREATE:$gameIdHex] createClientTextStoreWithData") {
|
||||
GamesManager.createClientTextStoreWithData(
|
||||
gameId = gameId,
|
||||
pregenerated = pregeneratedClientText,
|
||||
unrequestedTexts = unrequestedTexts,
|
||||
// FIXME: this is making it all accessible to everyone
|
||||
accessibleTo = unrequestedTexts.values
|
||||
.map(ut => ut.id -> initialEar.engine.factionIds)
|
||||
.toMap,
|
||||
persister = gamePersisterCreation.databaseUploadPersisterForGame(gameId)
|
||||
databasePersister = gamePersisterCreation.databaseUploadPersisterForGame(gameId)
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -112,6 +112,7 @@ class GamesManagerTest extends AnyFlatSpec with MockFactory with Matchers with B
|
||||
val dir = new File("/tmp/eagle0")
|
||||
|
||||
override def beforeEach(): Unit = {
|
||||
val _ = GamesManager.instanceStartTimeMs
|
||||
previousHistoryBackend = HistoryBackend.stringValue
|
||||
HistoryBackend.setStringValue("sqlite")
|
||||
mockRandom.reset()
|
||||
|
||||
Reference in New Issue
Block a user