Skip to content

Commit ad45919

Browse files
warcholjakubwarcholjakub
committed
migration script
1 parent 02be206 commit ad45919

3 files changed

Lines changed: 324 additions & 1 deletion

File tree

‎scripts/PostgresMigrationScript.scala‎

Lines changed: 317 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,317 @@
1+
//> using scala 3
2+
3+
//> using dep org.scastie.old::api::0.30.0-SNAPSHOT
4+
//> using dep org.scastie.old:storage_2.13:0.30.0-SNAPSHOT
5+
6+
//> using dep org.scastie:api_2.13:1.0.0-SNAPSHOT
7+
//> using dep org.scastie:storage_2.13:1.0.0-SNAPSHOT
8+
9+
//> using dep com.typesafe:config:latest.release
10+
//> using dep com.lihaoyi::pprint:latest.release
11+
12+
13+
14+
import org.scastie.storage.postgres.PostgresContainer
15+
import org.scastie.storage.postgres.Snippet
16+
17+
import org.slf4j.LoggerFactory
18+
import ch.qos.logback.classic.{Level, Logger}
19+
20+
import com.olegych.scastie.api.AttachedDom
21+
import com.olegych.scastie.api.Html
22+
import com.olegych.scastie.api.ScalaTarget.Js
23+
import com.olegych.scastie.api.ScalaTarget.Jvm
24+
import com.olegych.scastie.api.ScalaTarget.Native
25+
import com.olegych.scastie.api.ScalaTarget.Scala3
26+
import com.olegych.scastie.api.ScalaTarget.Typelevel
27+
import com.olegych.scastie.api.Value
28+
import com.olegych.scastie.{api => oldApi}
29+
import com.olegych.scastie.{storage => oldStorage}
30+
import com.typesafe.config.ConfigFactory
31+
import io.circe.Encoder
32+
import io.circe.*
33+
import org.mongodb.scala._
34+
import org.mongodb.scala.model.Projections._
35+
import org.mongodb.scala.model._
36+
import org.scastie.runtime.{api => runtimeApi}
37+
import org.scastie.storage.MongoSnippet.mongoSnippetEncoder
38+
import org.scastie.{api => newApi}
39+
import org.scastie.{storage => newStorage}
40+
import play.api.libs.json.Json
41+
import play.api.libs.json.Reads
42+
43+
import scala.concurrent.Await
44+
import scala.concurrent.ExecutionContext.Implicits.global
45+
import scala.concurrent.Future
46+
import scala.concurrent.duration.*
47+
import scala.util.Try
48+
49+
@main def migrate = {
50+
val rootLogger = LoggerFactory.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME).asInstanceOf[Logger]
51+
rootLogger.setLevel(Level.INFO)
52+
LoggerFactory.getLogger("com.zaxxer.hikari").asInstanceOf[Logger].setLevel(Level.WARN)
53+
LoggerFactory.getLogger("org.postgresql").asInstanceOf[Logger].setLevel(Level.WARN)
54+
LoggerFactory.getLogger("scalasql").asInstanceOf[Logger].setLevel(Level.WARN)
55+
LoggerFactory.getLogger("org.scastie.storage").asInstanceOf[Logger].setLevel(Level.WARN)
56+
System.setProperty("org.mongodb.driver.level", "WARN")
57+
val mongoUri = {
58+
val config = ConfigFactory.load().getConfig("scastie.mongodb")
59+
val user = config.getString("user")
60+
val password = config.getString("password")
61+
val databaseName = config.getString("database")
62+
val host = config.getString("host")
63+
val port = config.getInt("port")
64+
s"mongodb://$user:$password@$host:$port/$databaseName"
65+
// "mongodb://127.0.0.1:27017/?directConnection=true&serverSelectionTimeoutMS=2000&appName=mongosh+1.6.1"
66+
}
67+
val mongoClient: MongoClient =
68+
MongoClient(mongoUri)
69+
70+
val oldDatabase: MongoDatabase = mongoClient.getDatabase("snippets")
71+
val oldSnippets = oldDatabase.getCollection[Document]("snippets")
72+
73+
val pgContainer: PostgresContainer =
74+
new PostgresContainer(defaultConfig = true)
75+
76+
def fromBson[T](obj: Document)(implicit reads: Reads[T]): Option[T] =
77+
pprint(s"Deserializing BSON: ${obj.toJson().take(10)}")
78+
Json.parse(obj.toJson()).asOpt[T]
79+
80+
def convertType(o: oldApi.ProcessOutputType): newApi.ProcessOutputType =
81+
o match
82+
case oldApi.ProcessOutputType.StdErr => newApi.ProcessOutputType.StdErr
83+
case oldApi.ProcessOutputType.StdOut => newApi.ProcessOutputType.StdOut
84+
85+
def convertSeverity(o: oldApi.Severity): newApi.Severity =
86+
o match
87+
case oldApi.Info => newApi.Info
88+
case oldApi.Warning => newApi.Warning
89+
case oldApi.Error => newApi.Error
90+
91+
def convertInstrumentation(instrumentation: oldApi.Instrumentation): runtimeApi.Instrumentation =
92+
val newPosition = runtimeApi.Position(instrumentation.position.start, instrumentation.position.end)
93+
val newRender = instrumentation.render match
94+
case Value(v, className) => runtimeApi.Value(v, className)
95+
case Html(a, folded) => runtimeApi.Html(a, folded)
96+
case AttachedDom(uuid, folded) => runtimeApi.AttachedDom(uuid, folded)
97+
98+
runtimeApi.Instrumentation(newPosition, newRender)
99+
100+
def convertTarget(target: oldApi.ScalaTarget): newApi.SbtScalaTarget =
101+
target match
102+
case Jvm(scalaVersion) => newApi.Scala2(scalaVersion)
103+
case Typelevel(scalaVersion) => newApi.Typelevel(scalaVersion)
104+
case Js(scalaVersion, scalaJsVersion) => newApi.Js(scalaVersion, scalaJsVersion)
105+
case Native(scalaVersion, scalaNativeVersion) => newApi.Native(scalaVersion, scalaNativeVersion)
106+
case Scala3(scalaVersion) => newApi.Scala3(scalaVersion)
107+
108+
def convertLibrariesFromList(libs: List[(oldApi.ScalaDependency, oldApi.Project)], newTarget: newApi.SbtScalaTarget): List[(newApi.ScalaDependency, newApi.Project)] =
109+
libs.map { (l, pro) =>
110+
val newL = newApi.ScalaDependency(l.groupId, l.artifact, newTarget, l.version)
111+
val newPro = newApi.Project(pro.organization, pro.repository, pro.logo, pro.artifacts)
112+
113+
newL -> newPro
114+
}
115+
116+
def convertSnippetId(old: oldApi.SnippetId): newApi.SnippetId =
117+
val newUser: Option[newApi.SnippetUserPart] = old.user.map(user => newApi.SnippetUserPart(user.login, user.update))
118+
newApi.SnippetId(base64UUID = old.base64UUID, user = newUser)
119+
120+
def convertSnippet(old: oldStorage.MongoSnippet) =
121+
import org.scastie.storage.*
122+
val newCode = if old.inputs.code.contains("import com.olegych") && old.oldId != 0 then
123+
old.inputs.code.replace("import com.olegych", "import org.scastie")
124+
else
125+
old.inputs.code
126+
127+
val newProgresses = old.progresses.map(p =>
128+
newApi.SnippetProgress(
129+
ts = p.ts,
130+
id = p.id,
131+
snippetId = p.snippetId.map(convertSnippetId),
132+
userOutput = p.userOutput.map(o => newApi.ProcessOutput(o.line, tpe = convertType(o.tpe), id = o.id)),
133+
buildOutput = p.sbtOutput.map(o => newApi.ProcessOutput(o.line, tpe = convertType(o.tpe), id = o.id)),
134+
compilationInfos = p.compilationInfos.map(problem => newApi.Problem(convertSeverity(problem.severity), problem.line, problem.message)),
135+
instrumentations = p.instrumentations.map(convertInstrumentation),
136+
runtimeError = p.runtimeError.map(er => runtimeApi.RuntimeError(er.message, er.line, er.fullStack)),
137+
scalaJsContent = p.scalaJsContent,
138+
scalaJsSourceMapContent = p.scalaJsSourceMapContent,
139+
isDone = p.isDone,
140+
isTimeout = p.isTimeout,
141+
isSbtError = p.isSbtError,
142+
isForcedProgramMode = p.isForcedProgramMode
143+
)
144+
)
145+
146+
val newTarget = convertTarget(old.inputs.target)
147+
val newInputs: newApi.BaseInputs = newApi.SbtInputs(
148+
isWorksheetMode = old.inputs.isWorksheetMode,
149+
code = newCode,
150+
target = newTarget,
151+
libraries = old.inputs.libraries.map(l => newApi.ScalaDependency(l.groupId, l.artifact, newTarget, l.version)),
152+
librariesFromList = convertLibrariesFromList(old.inputs.librariesFromList, newTarget),
153+
sbtConfigExtra = old.inputs.sbtConfigExtra,
154+
sbtConfigSaved = old.inputs.sbtConfigSaved,
155+
sbtPluginsConfigExtra = old.inputs.sbtPluginsConfigExtra,
156+
sbtPluginsConfigSaved = old.inputs.sbtPluginsConfigSaved,
157+
isShowingInUserProfile = old.inputs.isShowingInUserProfile,
158+
forked = old.inputs.forked.map(convertSnippetId)
159+
)
160+
161+
Snippet(
162+
simpleSnippetId = old.simpleSnippetId,
163+
username = old.user,
164+
snippetId = convertSnippetId(old.snippetId),
165+
inputs = newInputs,
166+
progresses = newProgresses,
167+
scalaJsContent = old.scalaJsContent,
168+
scalaJsSourceMapContent = old.scalaJsSourceMapContent,
169+
time = old.time
170+
)
171+
def saveSnippetInNewDatabase(snippets: Seq[Snippet]): Future[Unit] = {
172+
// Przetwarzaj 10 snippetów równocześnie zamiast wszystkich sekwencyjnie
173+
Future.traverse(snippets.grouped(10).toSeq) { batch =>
174+
Future.traverse(batch) { s =>
175+
for {
176+
_ <- pgContainer.insertWithExistingId(s.snippetId, s.inputs)
177+
_ <- Future.sequence(s.progresses.map(pgContainer.appendOutput)) // równolegle zamiast .traverse
178+
} yield ()
179+
}
180+
}.map(_ => ())
181+
}
182+
183+
def migrateSnippets(originalSize: Int): Future[Seq[String]] = {
184+
pprint.pprintln(s"Starting migration of $originalSize snippets")
185+
186+
var count = 0L
187+
var progress = 0
188+
val allUrls = scala.collection.mutable.ArrayBuffer.empty[String]
189+
190+
def step(): Future[Unit] = {
191+
if (count >= originalSize) {
192+
Future.unit
193+
} else {
194+
oldSnippets
195+
.find()
196+
.skip(count.toInt)
197+
.limit(1000)
198+
.toFuture()
199+
.flatMap { docs =>
200+
docs.foreach { doc =>
201+
val sOpt = fromBson[oldStorage.MongoSnippet](doc)
202+
sOpt match {
203+
case Some(s) =>
204+
if (s.user.isEmpty) println(s"Skipping anonymous snippet ${s.snippetId.url}")
205+
else pprint.pprintln(s"Migrating snippet ${s.snippetId.url}")
206+
case None =>
207+
pprint.pprintln(s"Could not deserialize snippet: ${doc.toJson()}")
208+
}
209+
}
210+
211+
val snippets = docs
212+
.flatMap(fromBson[oldStorage.MongoSnippet])
213+
.flatMap { s =>
214+
if (s.user.isEmpty) None
215+
else Some(convertSnippet(s))
216+
}
217+
count += docs.size
218+
219+
allUrls ++= snippets.map(_.snippetId.url)
220+
saveSnippetInNewDatabase(snippets).flatMap(_ => step())
221+
}
222+
}
223+
}
224+
225+
step().map(_ => allUrls.toSeq)
226+
}
227+
228+
def verifySnippets(migratedSnippetUrls: Seq[String]): Future[Unit] = {
229+
pprint.pprintln("Verifying snippets...")
230+
231+
var count = 0L
232+
var progress = 0
233+
var errors = 0
234+
235+
def step(): Future[Unit] = {
236+
if (count >= migratedSnippetUrls.size) {
237+
pprint.pprintln(s"Verification complete! Total errors: $errors")
238+
Future.unit
239+
} else {
240+
val batch = migratedSnippetUrls.slice(count.toInt, (count + 100).toInt)
241+
242+
Future.traverse(batch) { url =>
243+
val snippetId = newApi.SnippetId.fromString(url)
244+
245+
for {
246+
oldOpt <- {
247+
oldSnippets
248+
.find(Filters.eq("simpleSnippetId", url))
249+
.headOption()
250+
.map(_.flatMap(fromBson[oldStorage.MongoSnippet]))
251+
}
252+
newOpt <- Future.successful(pgContainer.readPostgresSnippet(snippetId))
253+
} yield (url, oldOpt, newOpt)
254+
}.flatMap { results =>
255+
results.foreach { case (url, oldOpt, newOpt) =>
256+
(oldOpt, newOpt) match {
257+
case (Some(old), Some(newS)) =>
258+
val oldCode = old.inputs.code.replace("import com.olegych", "import org.scastie")
259+
val newCode = newS.inputs match {
260+
case sbt: newApi.SbtInputs => sbt.code
261+
case cli: newApi.ScalaCliInputs => cli.code
262+
}
263+
264+
if (oldCode != newCode) {
265+
errors += 1
266+
pprint.pprintln(s"❌ Snippet $url mismatch")
267+
} else {
268+
pprint.pprintln(s"✓ Snippet $url verified successfully")
269+
}
270+
271+
case (None, Some(_)) =>
272+
errors += 1
273+
pprint.pprintln(s"❌ Snippet $url missing in old database")
274+
275+
case (Some(_), None) =>
276+
errors += 1
277+
pprint.pprintln(s"❌ Snippet $url missing in new database")
278+
279+
case (None, None) =>
280+
errors += 1
281+
pprint.pprintln(s"❌ Snippet $url missing in both databases")
282+
}
283+
}
284+
285+
count += batch.size
286+
val cProgress = (count * 100) / migratedSnippetUrls.size
287+
if (cProgress > progress) {
288+
progress = cProgress.toInt
289+
pprint.pprintln(s"Verification Progress: [$cProgress% / 100%]")
290+
}
291+
292+
step()
293+
}
294+
}
295+
}
296+
297+
step()
298+
}
299+
300+
try {
301+
val snippetsNumber = Await.result(oldSnippets.countDocuments().head, Duration.Inf)
302+
303+
pprint.pprintln(s"Starting migration of $snippetsNumber snippets...")
304+
305+
val migrationFuture = for {
306+
migratedUrls <- migrateSnippets(snippetsNumber.toInt)
307+
_ <- verifySnippets(migratedUrls)
308+
} yield ()
309+
310+
Await.result(migrationFuture, Duration.Inf)
311+
312+
} catch {
313+
case t: Throwable => t.printStackTrace()
314+
} finally {
315+
mongoClient.close()
316+
}
317+
}

‎storage/src/main/scala/org/scastie/storage/postgres/PostgresContainer.scala‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,8 @@ import scalasql.dialects.PostgresDialect
1313
class PostgresContainer(
1414
defaultConfig: Boolean = false,
1515
useConnectionPool: Boolean = true,
16-
runMigrations: Boolean = true
16+
runMigrations: Boolean = true,
17+
maxPoolSize: Int = 20
1718
)(
1819
implicit val ec: ExecutionContext
1920
) extends PostgresSnippetsContainer {
@@ -50,6 +51,7 @@ class PostgresContainer(
5051
ds.setUsername(user)
5152
ds.setPassword(password)
5253
ds.setPoolName("PostgresPool")
54+
ds.setMaximumPoolSize(maxPoolSize)
5355
ds
5456
case false =>
5557
val ds = new PGSimpleDataSource()

‎storage/src/main/scala/org/scastie/storage/postgres/PostgresSnippetsContainer.scala‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,10 @@ trait PostgresSnippetsContainer extends SnippetsContainer with PostgresConverter
3838
Future.unit
3939
}
4040

41+
def insertWithExistingId(snippetId: SnippetId, inputs: BaseInputs): Future[Unit] = {
42+
insert(snippetId, inputs)
43+
}
44+
4145
private def readProgresses(snippetId: SnippetId): List[SnippetProgress] = {
4246
val query = PostgresProgresses.select
4347
.filter(_.snippetId === snippetId.url)

0 commit comments

Comments
 (0)